fix message impl ctor

Author: BewareMyPower

include/pulsar/Message.h

@@ -217,9 +217,6 @@ class PULSAR_PUBLIC Message {
     MessageImplPtr impl_;
 
     Message(MessageImplPtr& impl);
-    Message(const MessageId& messageId, proto::BrokerEntryMetadata& brokerEntryMetadata,
-            proto::MessageMetadata& metadata, SharedBuffer& payload);
-    /// Used for Batch Messages
     Message(const MessageId& messageId, proto::BrokerEntryMetadata& brokerEntryMetadata,
             proto::MessageMetadata& metadata, SharedBuffer& payload,
             proto::SingleMessageMetadata& singleMetadata, const std::shared_ptr<std::string>& topicName);

lib/Commands.cc

@@ -926,12 +926,13 @@ Message Commands::deSerializeSingleMessageInBatch(Message& batchedMessage, int32
     const MessageId& m = batchedMessage.impl_->messageId;
     auto messageId = MessageIdBuilder::from(m).batchIndex(batchIndex).batchSize(batchSize).build();
     auto batchedMessageId = std::make_shared<BatchedMessageIdImpl>(*(messageId.impl_), acker);
-    Message singleMessage(MessageId{batchedMessageId}, batchedMessage.impl_->brokerEntryMetadata,
-                          batchedMessage.impl_->metadata, payload, metadata,
-                          batchedMessage.impl_->topicName_);
-    singleMessage.impl_->cnx_ = batchedMessage.impl_->cnx_;
 
-    return singleMessage;
+    auto msgImpl = std::make_shared<MessageImpl>(messageId, batchedMessage.impl_->brokerEntryMetadata,
+                                                 batchedMessage.impl_->metadata, payload, metadata,
+                                                 batchedMessage.impl_->topicName_, false);
+    msgImpl->cnx_ = batchedMessage.impl_->cnx_;
+
+    return Message(msgImpl);
 }
 
 MessageIdImplPtr Commands::getMessageIdImpl(const MessageId& messageId) { return messageId.impl_; }

lib/ConsumerImpl.cc

@@ -583,7 +583,7 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
     }
 
     auto msgImpl =
-        std::make_shared<MessageImpl>(messageId, metadata, brokerEntryMetadata, payload, std::nullopt,
+        std::make_shared<MessageImpl>(messageId, brokerEntryMetadata, metadata, payload, std::nullopt,
                                       getTopicPtr(), decryptResult == CONSUME_ENCRYPTED);
     Message m(msgImpl);
     m.impl_->cnx_ = cnx.get();

lib/Message.cc

@@ -71,15 +71,6 @@ Message::Message() : impl_() {}
 
 Message::Message(MessageImplPtr& impl) : impl_(impl) {}
 
-Message::Message(const MessageId& messageId, proto::BrokerEntryMetadata& brokerEntryMetadata,
-                 proto::MessageMetadata& metadata, SharedBuffer& payload)
-    : impl_(std::make_shared<MessageImpl>()) {
-    impl_->messageId = messageId;
-    impl_->brokerEntryMetadata = brokerEntryMetadata;
-    impl_->metadata = metadata;
-    impl_->payload = payload;
-}
-
 Message::Message(const MessageId& messageID, proto::BrokerEntryMetadata& brokerEntryMetadata,
                  proto::MessageMetadata& metadata, SharedBuffer& payload,
                  proto::SingleMessageMetadata& singleMetadata, const std::shared_ptr<std::string>& topicName)

lib/MessageImpl.cc

@@ -20,13 +20,55 @@
 
 #include <utility>
 
+#include "PulsarApi.pb.h"
+
 namespace pulsar {
 
-MessageImpl::MessageImpl(const MessageId& messageId, const proto::MessageMetadata& metadata,
-                         const proto::BrokerEntryMetadata& brokerEntryMetadata, const SharedBuffer& payload,
+MessageImpl::MessageImpl(const MessageId& messageId, const proto::BrokerEntryMetadata& brokerEntryMetadata,
+                         const proto::MessageMetadata& metadata, const SharedBuffer& payload,
                          const optional<proto::SingleMessageMetadata>& singleMetadata,
                          const std::shared_ptr<std::string>& topicName, bool isDecryptionFailed)
-    : encryptionContext_(std::in_place, metadata, isDecryptionFailed) {}
+    : messageId(messageId),
+      brokerEntryMetadata(brokerEntryMetadata),
+      metadata(metadata),
+      payload(payload),
+      topicName_(topicName),
+      encryptionContext_(std::in_place, metadata, isDecryptionFailed) {
+    if (singleMetadata.has_value()) {
+        this->metadata.clear_properties();
+        if (singleMetadata->properties_size() > 0) {
+            this->metadata.mutable_properties()->Reserve(singleMetadata->properties_size());
+            for (int i = 0; i < singleMetadata->properties_size(); i++) {
+                auto keyValue = proto::KeyValue().New();
+                *keyValue = singleMetadata->properties(i);
+                this->metadata.mutable_properties()->AddAllocated(keyValue);
+            }
+        }
+        if (singleMetadata->has_partition_key()) {
+            this->metadata.set_partition_key(singleMetadata->partition_key());
+        } else {
+            this->metadata.clear_partition_key();
+        }
+
+        if (singleMetadata->has_ordering_key()) {
+            this->metadata.set_ordering_key(singleMetadata->ordering_key());
+        } else {
+            this->metadata.clear_ordering_key();
+        }
+
+        if (singleMetadata->has_event_time()) {
+            this->metadata.set_event_time(singleMetadata->event_time());
+        } else {
+            this->metadata.clear_event_time();
+        }
+
+        if (singleMetadata->has_sequence_id()) {
+            this->metadata.set_sequence_id(singleMetadata->sequence_id());
+        } else {
+            this->metadata.clear_sequence_id();
+        }
+    }
+}
 
 const Message::StringMap& MessageImpl::properties() {
     if (properties_.size() == 0) {

lib/MessageImpl.h

@@ -42,8 +42,8 @@ class MessageImpl {
    public:
     // TODO: remove this default constructor
     explicit MessageImpl() : encryptionContext_(std::nullopt) {}
-    MessageImpl(const MessageId& messageId, const proto::MessageMetadata& metadata,
-                const proto::BrokerEntryMetadata& brokerEntryMetadata, const SharedBuffer& payload,
+    MessageImpl(const MessageId& messageId, const proto::BrokerEntryMetadata& brokerEntryMetadata,
+                const proto::MessageMetadata& metadata, const SharedBuffer& payload,
                 const optional<proto::SingleMessageMetadata>& singleMetadata,
                 const std::shared_ptr<std::string>& topicName, bool undecryptedPayload);
 
@@ -81,11 +81,11 @@ class MessageImpl {
     friend class PulsarWrapper;
     friend class MessageBuilder;
 
+    MessageId messageId;
     proto::BrokerEntryMetadata brokerEntryMetadata;
     proto::MessageMetadata metadata;
     SharedBuffer payload;
     std::shared_ptr<KeyValueImpl> keyValuePtr;
-    MessageId messageId;
     ClientConnection* cnx_;
     std::shared_ptr<std::string> topicName_;
     int redeliveryCount_;