[API-340] Add documentation and code samples for the ReliableTopic

docs/features.rst

@@ -11,6 +11,7 @@ features:
 - MultiMap
 - Replicated Map
 - Ringbuffer
+- ReliableTopic
 - Topic
 - CRDT PN Counter
 - Flake Id Generator

docs/using_python_client_with_hazelcast_imdg.rst

@@ -356,6 +356,55 @@ A Ringbuffer usage example is shown below.
     sequence += 1
     print(ringbuffer.read_one(sequence))  # Outputs '200'
 
+Using ReliableTopic
+~~~~~~~~~~~~~~~~~~~
+
+Hazelcast ReliableTopic is a distributed topic implementation backed up by the Ringbuffer
+data structure. For details, see the
+`Reliable Topic section <https://docs.hazelcast.com/imdg/latest/data-structures/reliable-topic.html>`__
+in the Hazelcast IMDG Reference Manual.
+
+A Reliable Topic usage example is shown below.
+
+.. code:: python
+
+    # Get a Topic called "my-distributed-topic"
+    topic = client.get_reliable_topic("my-distributed-topic").blocking()
+
+    # Add a Listener to the Topic
+    topic.add_listener(lambda message: print(message))
+
+    # Publish a message to the Topic
+    topic.publish("Hello to distributed world")
+
+Configuring Reliable Topic
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+You may configure Reliable Topics using the ``reliable_topics``
+argument:
+
+.. code:: python
+
+    client = hazelcast.HazelcastClient(
+        reliable_topics={
+            "my-topic": {
+                "overload_policy": TopicOverloadPolicy.DISCARD_OLDEST,
+                "read_batch_size": 20,
+            }
+        }
+    )
+
+The following are the descriptions of configuration elements and
+attributes:
+
+- keys of the dictionary: Name of the Reliable Topic.
+- ``overload_policy``: Policy to handle an overloaded topic. By default,
+  set to ``BLOCK``.
+- ``read_batch_size``: Number of messages the reliable topic will try to
+  read in batch. It will get at least one, but if there are more
+  available, then it will try to get more to increase throughput.
+  By default, set to ``10``.
+
 Using Topic
 ~~~~~~~~~~~
 

examples/org-website/reliable_topic_sample.py

@@ -0,0 +1,16 @@
+import hazelcast
+
+# Start the Hazelcast Client and connect to an already running Hazelcast Cluster on 127.0.0.1
+client = hazelcast.HazelcastClient()
+
+# Get a Topic called "my-distributed-topic"
+topic = client.get_reliable_topic("my-distributed-topic").blocking()
+
+# Add a Listener to the Topic
+topic.add_listener(lambda message: print(message))
+
+# Publish a message to the Topic
+topic.publish("Hello to distributed world")
+
+# Shutdown this Hazelcast Client
+client.shutdown()

examples/reliable-topic/reliable_topic_example.py

@@ -0,0 +1,58 @@
+import hazelcast
+from hazelcast.config import TopicOverloadPolicy
+from hazelcast.proxy.reliable_topic import ReliableMessageListener
+
+# Customize the reliable topic
+client = hazelcast.HazelcastClient(
+    reliable_topics={
+        "my-topic": {
+            "overload_policy": TopicOverloadPolicy.DISCARD_OLDEST,
+            "read_batch_size": 20,
+        }
+    }
+)
+
+topic = client.get_reliable_topic("my-topic").blocking()
+
+# Add a listener with a callable
+reg_id = topic.add_listener(lambda m: print("First listener:", m))
+
+
+# Or, customize the behaviour of the listener
+# via ReliableMessageListener
+class MyListener(ReliableMessageListener):
+    def on_message(self, message):
+        print("Second listener:", message)
+
+    def retrieve_initial_sequence(self):
+        return 0
+
+    def store_sequence(self, sequence):
+        pass
+
+    def is_loss_tolerant(self):
+        return True
+
+    def is_terminal(self, error):
+        return False
+
+
+# Add a custom ReliableMessageListener
+topic.add_listener(MyListener())
+
+
+for i in range(100):
+    # Publish messages one-by-one
+    topic.publish(i)
+
+
+messages = range(100, 200)
+
+# Publish message in batch
+topic.publish_all(messages)
+
+# Remove listener so that it won't receive
+# messages anymore
+topic.remove_listener(reg_id)
+
+client.shutdown()

hazelcast/client.py

@@ -270,7 +270,7 @@ class SomeClassSerializer(StreamSerializer):
 
         reliable_topics (dict[str, dict[str, any]]): Dictionary of reliable
             topic names and the corresponding reliable topic configurations as
-            a dictionary. The reliable topic configurations contains the
+            a dictionary. The reliable topic configurations contain the
             following options. When an option is missing from the
             configuration, it will be set to its default value.
 

hazelcast/config.py

@@ -205,7 +205,7 @@ class ReconnectMode(object):
 
 
 class TopicOverloadPolicy(object):
-    """A policy to deal with an overloaded topic; so topic where there is no
+    """A policy to deal with an overloaded topic; a topic where there is no
     place to store new messages.
 
     The reliable topic uses a :class:`hazelcast.proxy.ringbuffer.Ringbuffer` to

hazelcast/proxy/base.py

@@ -306,6 +306,14 @@ def message(self):
         self._message = self._to_object(self._message_data)
         return self._message
 
+    def __repr__(self):
+        return "TopicMessage(message=%s, publish_time=%s, topic_name=%s, publishing_member=%s)" % (
+            self.message,
+            self.publish_time,
+            self.name,
+            self.member,
+        )
+
 
 def get_entry_listener_flags(**kwargs):
     flags = 0

hazelcast/proxy/reliable_topic.py

@@ -83,10 +83,9 @@ def on_message(self, message):
         One should not block in this callback. If blocking is necessary,
         consider delegating that task to an executor or a thread pool.
 
-
         Args:
             message (hazelcast.proxy.base.TopicMessage): The message that
-            is received for the topic
+                is received for the topic
         """
         raise NotImplementedError("on_message")
 
@@ -108,7 +107,7 @@ def retrieve_initial_sequence(self):
         raise NotImplementedError("retrieve_initial_sequence")
 
     def store_sequence(self, sequence):
-        """ "
+        """
         Informs the ReliableMessageListener that it should store the sequence.
         This method is called before the message is processed. Can be used to
         make a durable subscription.

tests/integration/backward_compatible/proxy/reliable_topic_test.py

@@ -117,11 +117,11 @@ def is_loss_tolerant(self):
             def is_terminal(self, error):
                 return False
 
+        topic.publish_all(range(10))
+
         registration_id = topic.add_listener(Listener())
         self.assertIsNotNone(registration_id)
 
-        topic.publish_all(range(10))
-
         self.assertTrueEventually(lambda: self.assertEqual(list(range(5, 10)), messages))
 
     def test_add_listener_with_store_sequence(self):

tests/integration/backward_compatible/ssl_tests/ssl_test.py

@@ -65,9 +65,9 @@ def test_ssl_enabled_map_size(self):
                 cluster.id, True, get_abs_path(self.current_directory, "server1-cert.pem")
             )
         )
-        test_map = client.get_map("test_map")
+        test_map = client.get_map("test_map").blocking()
         fill_map(test_map, 10)
-        self.assertEqual(test_map.size().result(), 10)
+        self.assertEqual(test_map.size(), 10)
         client.shutdown()
 
     def test_ssl_enabled_with_custom_ciphers(self):