[pulsar-client-cpp] Add MessageId to send Message (apache#7439)

* fix: add message id to send message in cpp * test: fix produce consume test to check whether message id was changed or not
blitznihar · Jul 5, 2020 · ca783fc · ca783fc
1 parent 35bb235
commit ca783fc
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 0 deletions.
diff --git a/pulsar-client-cpp/include/pulsar/Message.h b/pulsar-client-cpp/include/pulsar/Message.h
@@ -103,6 +103,12 @@ class PULSAR_PUBLIC Message {
      */
     const MessageId& getMessageId() const;
 
+    /**
+     * Set the unique message ID.
+     *
+     */
+    void setMessageId(const MessageId& messageID) const;
+
     /**
      * Get the partition key for this message
      * @return key string that is hashed to determine message's topic partition

diff --git a/pulsar-client-cpp/lib/Message.cc b/pulsar-client-cpp/lib/Message.cc
@@ -96,6 +96,13 @@ const MessageId& Message::getMessageId() const {
     }
 }
 
+void Message::setMessageId(const MessageId& messageID) const {
+    if (impl_) {
+        impl_->messageId = messageID;
+    }
+    return;
+}
+
 bool Message::hasPartitionKey() const {
     if (impl_) {
         return impl_->hasPartitionKey();

diff --git a/pulsar-client-cpp/lib/Producer.cc b/pulsar-client-cpp/lib/Producer.cc
@@ -44,6 +44,8 @@ Result Producer::send(const Message& msg) {
 
     MessageId mi;
     Result result = promise.getFuture().get(mi);
+    msg.setMessageId(mi);
+
     return result;
 }
 

diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
@@ -268,8 +268,10 @@ TEST(BasicEndToEndTest, testProduceConsume) {
     // Send synchronously
     std::string content = "msg-1-content";
     Message msg = MessageBuilder().setContent(content).build();
+    ASSERT_EQ(MessageId(-1, -1, -1, -1), msg.getMessageId());
     result = producer.send(msg);
     ASSERT_EQ(ResultOk, result);
+    ASSERT_NE(MessageId(-1, -1, -1, -1), msg.getMessageId());
 
     Message receivedMsg;
     consumer.receive(receivedMsg);