publish_all() method for Topic [API-1529]

examples/topic/topic_example.py

@@ -16,4 +16,7 @@ def on_message(event):
  topic.publish("Message " + str(i))
  time.sleep(0.1)
 
+topic.publish_all(["m1", "m2", "m3", "m4", "m5"])
+time.sleep(1)
+
 client.shutdown()

hazelcast/protocol/codec/topic_publish_all_codec.py

@@ -0,0 +1,18 @@
+from hazelcast.protocol.client_message import OutboundMessage, REQUEST_HEADER_SIZE, create_initial_buffer
+from hazelcast.protocol.builtin import StringCodec
+from hazelcast.protocol.builtin import ListMultiFrameCodec
+from hazelcast.protocol.builtin import DataCodec
+
+# hex: 0x040400
+_REQUEST_MESSAGE_TYPE = 263168
+# hex: 0x040401
+_RESPONSE_MESSAGE_TYPE = 263169
+
+_REQUEST_INITIAL_FRAME_SIZE = REQUEST_HEADER_SIZE
+
+
+def encode_request(name, messages):
+ buf = create_initial_buffer(_REQUEST_INITIAL_FRAME_SIZE, _REQUEST_MESSAGE_TYPE)
+ StringCodec.encode(buf, name)
+ ListMultiFrameCodec.encode(buf, messages, DataCodec.encode, True)
+ return OutboundMessage(buf, False)

hazelcast/proxy/topic.py

@@ -1,9 +1,11 @@
 import typing
 
+from hazelcast.errors import NullPointerError
 from hazelcast.future import Future
 from hazelcast.protocol.codec import (
  topic_add_message_listener_codec,
  topic_publish_codec,
+ topic_publish_all_codec,
  topic_remove_message_listener_codec,
 )
 from hazelcast.proxy.base import PartitionSpecificProxy, TopicMessage
@@ -57,7 +59,7 @@ def handle(item_data, publish_time, uuid):
  )
 
  def publish(self, message: MessageType) -> Future[None]:
- """Publishes the message to all subscribers of this topic
+ """Publishes the message to all subscribers of this topic.
 
  Args:
  message: The message to be published.
@@ -70,6 +72,27 @@ def publish(self, message: MessageType) -> Future[None]:
  request = topic_publish_codec.encode_request(self.name, message_data)
  return self._invoke(request)
 
+ def publish_all(self, messages: typing.Sequence[MessageType]) -> Future[None]:
+ """Publishes the messages to all subscribers of this topic.
+
+ Args:
+ messages: The messages to be published.
+ """
+ try:
+ if messages is None:
+ raise NullPointerError("Null message is not allowed!")
+ data_list = []
+ for m in messages:
+ data = self._to_data(m)
+ if data is None:
+ raise NullPointerError("Null message is not allowed!")
+ data_list.append(data)
+ except SchemaNotReplicatedError as e:
+ return self._send_schema_and_retry(e, self.publish_all, messages)
+
+ request = topic_publish_all_codec.encode_request(self.name, data_list)
+ return self._invoke(request)
+
  def remove_listener(self, registration_id: str) -> Future[bool]:
  """Stops receiving messages for the given message listener.
 
@@ -107,6 +130,12 @@ def publish( # type: ignore[override]
  ) -> None:
  return self._wrapped.publish(message).result()
 
+ def publish_all( # type: ignore[override]
+ self,
+ messages: typing.Sequence[MessageType]
+ ) -> None:
+ return self._wrapped.publish_all(messages).result()
+
  def remove_listener( # type: ignore[override]
  self,
  registration_id: str,

tests/integration/backward_compatible/proxy/topic_test.py

@@ -1,3 +1,4 @@
+from hazelcast.errors import NullPointerError
 from tests.base import SingleMemberTestCase
 from tests.util import random_string, event_collector
 
@@ -44,3 +45,24 @@ def assert_event():
 
  def test_str(self):
  self.assertTrue(str(self.topic).startswith("Topic"))
+
+ def test_publish_all(self):
+ collector = event_collector()
+ self.topic.add_listener(on_message=collector)
+
+ messages = ["message1", "message2", "message3"]
+ self.topic.publish_all(messages)
+
+ def assert_event():
+ self.assertEqual(len(collector.events), 3)
+
+ self.assertTrueEventually(assert_event, 5)
+
+ def test_publish_all_none_argument(self):
+ with self.assertRaises(NullPointerError):
+ self.topic.publish_all(None)
+
+ def test_publish_all_none_message(self):
+ messages = ["message1", None, "message3"]
+ with self.assertRaises(NullPointerError):
+ self.topic.publish_all(messages)