finish impl

Author: BewareMyPower

lib/ConsumerImpl.cc

@@ -582,10 +582,11 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
         }
     }
 
-    // TODO: create encryption context
-    Message m(messageId, brokerEntryMetadata, metadata, payload);
+    auto msgImpl =
+        std::make_shared<MessageImpl>(messageId, metadata, brokerEntryMetadata, payload, std::nullopt,
+                                      getTopicPtr(), decryptResult == CONSUME_ENCRYPTED);
+    Message m(msgImpl);
     m.impl_->cnx_ = cnx.get();
-    m.impl_->setTopicName(getTopicPtr());
     m.impl_->setRedeliveryCount(msg.redelivery_count());
 
     if (metadata.has_schema_version()) {

lib/MessageImpl.cc

@@ -17,17 +17,16 @@
  * under the License.
  */
 #include "MessageImpl.h"
+
 #include <utility>
 
 namespace pulsar {
 
 MessageImpl::MessageImpl(const MessageId& messageId, const proto::MessageMetadata& metadata,
                          const proto::BrokerEntryMetadata& brokerEntryMetadata, const SharedBuffer& payload,
                          const optional<proto::SingleMessageMetadata>& singleMetadata,
-                         const std::shared_ptr<std::string>& topicName, bool undecryptedPayload)
-    : encryptionContext_(std::in_place, metadata, undecryptedPayload) {
-    // TODO:
-}
+                         const std::shared_ptr<std::string>& topicName, bool isDecryptionFailed)
+    : encryptionContext_(std::in_place, metadata, isDecryptionFailed) {}
 
 const Message::StringMap& MessageImpl::properties() {
     if (properties_.size() == 0) {