Revert "Initial commit"

This reverts commit e8e3e23.

Author: BewareMyPower

include/pulsar/EncryptionContext.h

@@ -19,18 +19,22 @@
 #ifndef MESSAGE_HPP_
 #define MESSAGE_HPP_
 
-#include <pulsar/EncryptionContext.h>
 #include <pulsar/defines.h>
 
 #include <map>
 #include <memory>
-#include <optional>
 #include <string>
 
 #include "KeyValue.h"
 #include "MessageId.h"
 
 namespace pulsar {
+namespace proto {
+class CommandMessage;
+class BrokerEntryMetadata;
+class MessageMetadata;
+class SingleMessageMetadata;
+}  // namespace proto
 
 class SharedBuffer;
 class MessageBuilder;
@@ -198,19 +202,19 @@ class PULSAR_PUBLIC Message {
      */
     const std::string& getProducerName() const noexcept;
 
-    /**
-     * @return the optional encryption context that is present when the message is encrypted, the pointer is
-     * valid as the Message instance is alive
-     */
-    std::optional<const EncryptionContext*> getEncryptionContext() const;
-
     bool operator==(const Message& msg) const;
 
    protected:
     typedef std::shared_ptr<MessageImpl> MessageImplPtr;
     MessageImplPtr impl_;
 
-    Message(const 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);
     friend class PartitionedProducerImpl;
     friend class MultiTopicsConsumerImpl;
     friend class MessageBuilder;

include/pulsar/Message.h

@@ -906,8 +906,7 @@ uint64_t Commands::serializeSingleMessagesToBatchPayload(SharedBuffer& batchPayl
 }
 
 Message Commands::deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex,
-                                                  int32_t batchSize, const BatchMessageAckerPtr& acker,
-                                                  const optional<EncryptionContext>& encryptionContext) {
+                                                  int32_t batchSize, const BatchMessageAckerPtr& acker) {
     SharedBuffer& uncompressedPayload = batchedMessage.impl_->payload;
 
     // Format of batch message
@@ -927,13 +926,12 @@ 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_;
 
-    auto msgImpl = std::make_shared<MessageImpl>(messageId, batchedMessage.impl_->brokerEntryMetadata,
-                                                 batchedMessage.impl_->metadata, payload, metadata,
-                                                 batchedMessage.impl_->topicName_, encryptionContext);
-    msgImpl->cnx_ = batchedMessage.impl_->cnx_;
-
-    return Message(msgImpl);
+    return singleMessage;
 }
 
 MessageIdImplPtr Commands::getMessageIdImpl(const MessageId& messageId) { return messageId.impl_; }

lib/Commands.cc

@@ -155,8 +155,7 @@ class Commands {
                                                           const std::vector<Message>& messages);
 
     static Message deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex,
-                                                   int32_t batchSize, const BatchMessageAckerPtr& acker,
-                                                   const optional<EncryptionContext>& encryptionContext);
+                                                   int32_t batchSize, const BatchMessageAckerPtr& acker);
 
     static MessageIdImplPtr getMessageIdImpl(const MessageId& messageId);
 

lib/Commands.h

@@ -548,27 +548,25 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
                                    bool& isChecksumValid, proto::BrokerEntryMetadata& brokerEntryMetadata,
                                    proto::MessageMetadata& metadata, SharedBuffer& payload) {
     LOG_DEBUG(getName() << "Received Message -- Size: " << payload.readableBytes());
-    if (!isChecksumValid) {
-        // Message discarded for checksum error
-        discardCorruptedMessage(cnx, msg.message_id(), CommandAck_ValidationError_ChecksumMismatch);
+
+    if (!decryptMessageIfNeeded(cnx, msg, metadata, payload)) {
+        // Message was discarded or not consumed due to decryption failure
         return;
     }
 
-    auto encryptionContext = metadata.encryption_keys_size() > 0
-                                 ? optional<EncryptionContext>{std::in_place, metadata, false}
-                                 : std::nullopt;
-
-    auto decryptResult = decryptMessageIfNeeded(cnx, encryptionContext, payload, msg.message_id());
-    if (decryptResult == FAILED) {
-        // Message was discarded due to decryption failure or not consumed due to decryption failure
+    if (!isChecksumValid) {
+        // Message discarded for checksum error
+        discardCorruptedMessage(cnx, msg.message_id(), CommandAck_ValidationError_ChecksumMismatch);
         return;
-    } else if (decryptResult == CONSUME_ENCRYPTED) {
-        encryptionContext->isDecryptionFailed_ = true;
     }
 
     auto redeliveryCount = msg.redelivery_count();
+    const bool isMessageUndecryptable =
+        metadata.encryption_keys_size() > 0 && !config_.getCryptoKeyReader().get() &&
+        config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::CONSUME;
+
     const bool isChunkedMessage = metadata.num_chunks_from_msg() > 1;
-    if (decryptResult == DECRYPTED && !isChunkedMessage) {
+    if (!isMessageUndecryptable && !isChunkedMessage) {
         if (!uncompressMessageIfNeeded(cnx, msg.message_id(), metadata, payload, true)) {
             // Message was discarded on decompression error
             return;
@@ -588,9 +586,9 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
         }
     }
 
-    Message m{std::make_shared<MessageImpl>(messageId, brokerEntryMetadata, metadata, payload, std::nullopt,
-                                            getTopicPtr(), std::move(encryptionContext))};
+    Message m(messageId, brokerEntryMetadata, metadata, payload);
     m.impl_->cnx_ = cnx.get();
+    m.impl_->setTopicName(getTopicPtr());
     m.impl_->setRedeliveryCount(msg.redelivery_count());
 
     if (metadata.has_schema_version()) {
@@ -612,16 +610,14 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
         return;
     }
 
-    // When the decryption failed, the whole batch message will be treated as a single message.
-    if (metadata.has_num_messages_in_batch() && decryptResult == DECRYPTED) {
+    if (metadata.has_num_messages_in_batch()) {
         BitSet::Data words(msg.ack_set_size());
         for (int i = 0; i < words.size(); i++) {
             words[i] = msg.ack_set(i);
         }
         BitSet ackSet{std::move(words)};
         Lock lock(mutex_);
-        numOfMessageReceived =
-            receiveIndividualMessagesFromBatch(cnx, m, ackSet, msg.redelivery_count(), encryptionContext);
+        numOfMessageReceived = receiveIndividualMessagesFromBatch(cnx, m, ackSet, msg.redelivery_count());
     } else {
         // try convert key value data.
         m.impl_->convertPayloadToKeyValue(config_.getSchema());
@@ -746,9 +742,9 @@ void ConsumerImpl::notifyPendingReceivedCallback(Result result, Message& msg,
 }
 
 // Zero Queue size is not supported with Batch Messages
-uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(
-    const ClientConnectionPtr& cnx, Message& batchedMessage, const BitSet& ackSet, int redeliveryCount,
-    const optional<EncryptionContext>& encryptionContext) {
+uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnectionPtr& cnx,
+                                                          Message& batchedMessage, const BitSet& ackSet,
+                                                          int redeliveryCount) {
     auto batchSize = batchedMessage.impl_->metadata.num_messages_in_batch();
     LOG_DEBUG("Received Batch messages of size - " << batchSize
                                                    << " -- msgId: " << batchedMessage.getMessageId());
@@ -760,8 +756,7 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(
     std::vector<Message> possibleToDeadLetter;
     for (int i = 0; i < batchSize; i++) {
         // This is a cheap copy since message contains only one shared pointer (impl_)
-        Message msg =
-            Commands::deSerializeSingleMessageInBatch(batchedMessage, i, batchSize, acker, encryptionContext);
+        Message msg = Commands::deSerializeSingleMessageInBatch(batchedMessage, i, batchSize, acker);
         msg.impl_->setRedeliveryCount(redeliveryCount);
         msg.impl_->setTopicName(batchedMessage.impl_->topicName_);
         msg.impl_->convertPayloadToKeyValue(config_.getSchema());
@@ -817,51 +812,50 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(
     return batchSize - skippedMessages;
 }
 
-auto ConsumerImpl::decryptMessageIfNeeded(const ClientConnectionPtr& cnx,
-                                          const optional<EncryptionContext>& context, SharedBuffer& payload,
-                                          const proto::MessageIdData& msgId) -> DecryptResult {
-    if (!context.has_value()) {
-        return DECRYPTED;
+bool ConsumerImpl::decryptMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
+                                          const proto::MessageMetadata& metadata, SharedBuffer& payload) {
+    if (!metadata.encryption_keys_size()) {
+        return true;
     }
 
     // If KeyReader is not configured throw exception based on config param
     if (!config_.isEncryptionEnabled()) {
         if (config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::CONSUME) {
             LOG_WARN(getName() << "CryptoKeyReader is not implemented. Consuming encrypted message.");
-            return CONSUME_ENCRYPTED;
+            return true;
         } else if (config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::DISCARD) {
             LOG_WARN(getName() << "Skipping decryption since CryptoKeyReader is not implemented and config "
                                   "is set to discard");
-            discardCorruptedMessage(cnx, msgId, CommandAck_ValidationError_DecryptionError);
+            discardCorruptedMessage(cnx, msg.message_id(), CommandAck_ValidationError_DecryptionError);
         } else {
             LOG_ERROR(getName() << "Message delivery failed since CryptoKeyReader is not implemented to "
                                    "consume encrypted message");
-            auto messageId = MessageIdBuilder::from(msgId).build();
+            auto messageId = MessageIdBuilder::from(msg.message_id()).build();
             unAckedMessageTrackerPtr_->add(messageId);
         }
-        return FAILED;
+        return false;
     }
 
     SharedBuffer decryptedPayload;
-    if (msgCrypto_->decrypt(*context, payload, config_.getCryptoKeyReader(), decryptedPayload)) {
+    if (msgCrypto_->decrypt(metadata, payload, config_.getCryptoKeyReader(), decryptedPayload)) {
         payload = decryptedPayload;
-        return DECRYPTED;
+        return true;
     }
 
     if (config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::CONSUME) {
         // Note, batch message will fail to consume even if config is set to consume
         LOG_WARN(
             getName() << "Decryption failed. Consuming encrypted message since config is set to consume.");
-        return CONSUME_ENCRYPTED;
+        return true;
     } else if (config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::DISCARD) {
         LOG_WARN(getName() << "Discarding message since decryption failed and config is set to discard");
-        discardCorruptedMessage(cnx, msgId, CommandAck_ValidationError_DecryptionError);
+        discardCorruptedMessage(cnx, msg.message_id(), CommandAck_ValidationError_DecryptionError);
     } else {
         LOG_ERROR(getName() << "Message delivery failed since unable to decrypt incoming message");
-        auto messageId = MessageIdBuilder::from(msgId).build();
+        auto messageId = MessageIdBuilder::from(msg.message_id()).build();
         unAckedMessageTrackerPtr_->add(messageId);
     }
-    return FAILED;
+    return false;
 }
 
 bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx,