Support deserializing a message id from bytes and topic (#259)

Author: BewareMyPower

pulsar/__init__.py

@@ -131,12 +131,30 @@ def __gt__(self, other) -> bool:
         return self._msg_id > other._msg_id
 
     @staticmethod
-    def deserialize(message_id_bytes):
+    def deserialize(message_id_bytes, topic: Optional[str] = None) -> _pulsar.MessageId:
         """
         Deserialize a message id object from a previously
         serialized bytes sequence.
+
+        Parameters
+        ----------
+        topic: str, optional
+            For multi-topics consumers, the topic name is required to deserialize the message id.
+
+            .. code-block:: python
+
+                msg = consumer.receive()
+                topic = msg.topic_name()
+                msg_id_bytes = msg.message_id().serialize()
+                # Store topic and msg_id_bytes somewhere
+                # Later, deserialize the message id
+                msg_id = MessageId.deserialize(msg_id_bytes, topic=topic)
+
         """
-        return _pulsar.MessageId.deserialize(message_id_bytes)
+        msg_id = _pulsar.MessageId.deserialize(message_id_bytes)
+        if topic is not None:
+            msg_id.topic_name(topic)
+        return msg_id
 
     @classmethod
     def wrap(cls, msg_id: _pulsar.MessageId):

src/message.cc

@@ -72,6 +72,10 @@ void export_message(py::module_& m) {
         .def("entry_id", &MessageId::entryId)
         .def("batch_index", &MessageId::batchIndex)
         .def("partition", &MessageId::partition)
+        .def(
+            "topic_name",
+            [](MessageId& msgId, const std::string& topicName) { msgId.setTopicName(topicName); },
+            return_value_policy::copy)
         .def_property_readonly_static("earliest", [](object) { return MessageId::earliest(); })
         .def_property_readonly_static("latest", [](object) { return MessageId::latest(); })
         .def("serialize",

tests/pulsar_test.py

@@ -1994,6 +1994,30 @@ def test_consumer_name(self):
         self.assertEqual(consumer.consumer_name(), name)
         client.close()
 
+    def test_deserialize_msg_id_with_topic(self):
+        client = Client(self.serviceUrl)
+        topic1 = "deserialize-msg-id-with-topic1-" + str(time.time())
+        topic2 = "deserialize-msg-id-with-topic2-" + str(time.time())
+        consumer = client.subscribe([topic1, topic2], 'sub')
+        producer1 = client.create_producer(topic1)
+        producer2 = client.create_producer(topic2)
+        producer1.send(b"msg-1")
+        producer2.send(b"msg-2")
+
+        serialized_msg_ids = dict()
+        for _ in range(2):
+            msg = consumer.receive(TM)
+            serialized_msg_ids[msg.topic_name()] = msg.message_id().serialize()
+        for topic, serialized_msg_id in serialized_msg_ids.items():
+            deserialized_msg_id = MessageId.deserialize(serialized_msg_id, topic=topic)
+            consumer.acknowledge_cumulative(deserialized_msg_id)
+        consumer.close()
+
+        consumer = client.subscribe([topic1, topic2], 'sub')
+        producer1.send(b'msg-3')
+        msg = consumer.receive(TM)
+        self.assertEqual(msg.value(), b'msg-3')
+        client.close()
 
 if __name__ == "__main__":
     main()