| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| #include "Commands.h" |
| |
| #include <pulsar/MessageBuilder.h> |
| #include <pulsar/MessageIdBuilder.h> |
| #include <pulsar/Schema.h> |
| #include <pulsar/Version.h> |
| |
| #include <algorithm> |
| #include <mutex> |
| |
| #include "BatchMessageAcker.h" |
| #include "BatchedMessageIdImpl.h" |
| #include "BitSet.h" |
| #include "ChunkMessageIdImpl.h" |
| #include "LogUtils.h" |
| #include "MessageImpl.h" |
| #include "PulsarApi.pb.h" |
| #include "Url.h" |
| #include "checksum/ChecksumProvider.h" |
| |
| using namespace pulsar; |
| namespace pulsar { |
| |
| DECLARE_LOG_OBJECT(); |
| |
| using proto::AuthData; |
| using proto::BaseCommand; |
| using proto::CommandAck; |
| using proto::CommandAuthResponse; |
| using proto::CommandCloseConsumer; |
| using proto::CommandCloseProducer; |
| using proto::CommandConnect; |
| using proto::CommandConsumerStats; |
| using proto::CommandFlow; |
| using proto::CommandGetLastMessageId; |
| using proto::CommandGetTopicsOfNamespace; |
| using proto::CommandLookupTopic; |
| using proto::CommandPartitionedTopicMetadata; |
| using proto::CommandProducer; |
| using proto::CommandRedeliverUnacknowledgedMessages; |
| using proto::CommandSeek; |
| using proto::CommandSend; |
| using proto::CommandSubscribe; |
| using proto::CommandUnsubscribe; |
| using proto::FeatureFlags; |
| using proto::IntRange; |
| using proto::KeySharedMeta; |
| using proto::MessageIdData; |
| using proto::ProtocolVersion_MAX; |
| using proto::SingleMessageMetadata; |
| |
| static inline bool isBuiltInSchema(SchemaType schemaType) { |
| switch (schemaType) { |
| case STRING: |
| case JSON: |
| case AVRO: |
| case PROTOBUF: |
| case PROTOBUF_NATIVE: |
| case KEY_VALUE: |
| return true; |
| |
| default: |
| return false; |
| } |
| } |
| |
| static inline proto::Schema_Type getSchemaType(SchemaType type) { |
| switch (type) { |
| case SchemaType::NONE: |
| return proto::Schema_Type_None; |
| case STRING: |
| return proto::Schema_Type_String; |
| case JSON: |
| return proto::Schema_Type_Json; |
| case PROTOBUF: |
| return proto::Schema_Type_Protobuf; |
| case AVRO: |
| return proto::Schema_Type_Avro; |
| case PROTOBUF_NATIVE: |
| return proto::Schema_Type_ProtobufNative; |
| case KEY_VALUE: |
| return proto::Schema_Type_KeyValue; |
| default: |
| return proto::Schema_Type_None; |
| } |
| } |
| |
| static proto::Schema* getSchema(const SchemaInfo& schemaInfo) { |
| proto::Schema* schema = proto::Schema().New(); |
| schema->set_name(schemaInfo.getName()); |
| schema->set_schema_data(schemaInfo.getSchema()); |
| schema->set_type(getSchemaType(schemaInfo.getSchemaType())); |
| for (const auto& kv : schemaInfo.getProperties()) { |
| proto::KeyValue* keyValue = proto::KeyValue().New(); |
| keyValue->set_key(kv.first); |
| keyValue->set_value(kv.second); |
| schema->mutable_properties()->AddAllocated(keyValue); |
| } |
| |
| return schema; |
| } |
| |
| SharedBuffer Commands::writeMessageWithSize(const BaseCommand& cmd) { |
| size_t cmdSize = cmd.ByteSize(); |
| size_t frameSize = 4 + cmdSize; |
| size_t bufferSize = 4 + frameSize; |
| |
| SharedBuffer buffer = SharedBuffer::allocate(bufferSize); |
| |
| buffer.writeUnsignedInt(frameSize); |
| buffer.writeUnsignedInt(cmdSize); |
| cmd.SerializeToArray(buffer.mutableData(), cmdSize); |
| buffer.bytesWritten(cmdSize); |
| return buffer; |
| } |
| |
| SharedBuffer Commands::newPartitionMetadataRequest(const std::string& topic, uint64_t requestId) { |
| static BaseCommand cmd; |
| static std::mutex mutex; |
| std::lock_guard<std::mutex> lock(mutex); |
| cmd.set_type(BaseCommand::PARTITIONED_METADATA); |
| CommandPartitionedTopicMetadata* partitionMetadata = cmd.mutable_partitionmetadata(); |
| partitionMetadata->set_topic(topic); |
| partitionMetadata->set_request_id(requestId); |
| const SharedBuffer buffer = writeMessageWithSize(cmd); |
| cmd.clear_partitionmetadata(); |
| return buffer; |
| } |
| |
| SharedBuffer Commands::newLookup(const std::string& topic, const bool authoritative, uint64_t requestId, |
| const std::string& listenerName) { |
| static BaseCommand cmd; |
| static std::mutex mutex; |
| std::lock_guard<std::mutex> lock(mutex); |
| cmd.set_type(BaseCommand::LOOKUP); |
| CommandLookupTopic* lookup = cmd.mutable_lookuptopic(); |
| lookup->set_topic(topic); |
| lookup->set_authoritative(authoritative); |
| lookup->set_request_id(requestId); |
| lookup->set_advertised_listener_name(listenerName); |
| const SharedBuffer buffer = writeMessageWithSize(cmd); |
| cmd.clear_lookuptopic(); |
| return buffer; |
| } |
| |
| SharedBuffer Commands::newGetSchema(const std::string& topic, const std::string& version, |
| uint64_t requestId) { |
| static BaseCommand cmd; |
| static std::mutex mutex; |
| std::lock_guard<std::mutex> lock(mutex); |
| cmd.set_type(BaseCommand::GET_SCHEMA); |
| |
| auto getSchema = cmd.mutable_getschema(); |
| getSchema->set_topic(topic); |
| getSchema->set_request_id(requestId); |
| if (!version.empty()) { |
| getSchema->set_schema_version(version); |
| } |
| |
| const SharedBuffer buffer = writeMessageWithSize(cmd); |
| cmd.clear_getschema(); |
| return buffer; |
| } |
| |
| SharedBuffer Commands::newConsumerStats(uint64_t consumerId, uint64_t requestId) { |
| static BaseCommand cmd; |
| static std::mutex mutex; |
| std::lock_guard<std::mutex> lock(mutex); |
| cmd.set_type(BaseCommand::CONSUMER_STATS); |
| CommandConsumerStats* consumerStats = cmd.mutable_consumerstats(); |
| consumerStats->set_consumer_id(consumerId); |
| consumerStats->set_request_id(requestId); |
| const SharedBuffer buffer = writeMessageWithSize(cmd); |
| cmd.clear_consumerstats(); |
| return buffer; |
| } |
| |
| PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint64_t producerId, |
| uint64_t sequenceId, ChecksumType checksumType, |
| const proto::MessageMetadata& metadata, const SharedBuffer& payload) { |
| cmd.set_type(BaseCommand::SEND); |
| CommandSend* send = cmd.mutable_send(); |
| send->set_producer_id(producerId); |
| send->set_sequence_id(sequenceId); |
| if (metadata.has_num_messages_in_batch()) { |
| send->set_num_messages(metadata.num_messages_in_batch()); |
| } |
| if (metadata.has_chunk_id()) { |
| send->set_is_chunk(true); |
| } |
| |
| // / Wire format |
| // [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD] |
| |
| int cmdSize = cmd.ByteSize(); |
| int msgMetadataSize = metadata.ByteSize(); |
| int payloadSize = payload.readableBytes(); |
| |
| int magicAndChecksumLength = (Crc32c == (checksumType)) ? (2 + 4 /* magic + checksumLength*/) : 0; |
| bool includeChecksum = magicAndChecksumLength > 0; |
| int headerContentSize = |
| 4 + cmdSize + magicAndChecksumLength + 4 + msgMetadataSize; // cmdLength + cmdSize + magicLength + |
| // checksumSize + msgMetadataLength + msgMetadataSize |
| int totalSize = headerContentSize + payloadSize; |
| int checksumReaderIndex = -1; |
| |
| headers.reset(); |
| assert(headers.writableBytes() >= (4 + headerContentSize)); // totalSize + headerLength |
| headers.writeUnsignedInt(totalSize); // External frame |
| |
| // Write cmd |
| headers.writeUnsignedInt(cmdSize); |
| cmd.SerializeToArray(headers.mutableData(), cmdSize); |
| headers.bytesWritten(cmdSize); |
| |
| // Create checksum placeholder |
| if (includeChecksum) { |
| headers.writeUnsignedShort(magicCrc32c); |
| checksumReaderIndex = headers.writerIndex(); |
| headers.skipBytes(checksumSize); // skip 4 bytes of checksum |
| } |
| |
| // Write metadata |
| headers.writeUnsignedInt(msgMetadataSize); |
| metadata.SerializeToArray(headers.mutableData(), msgMetadataSize); |
| headers.bytesWritten(msgMetadataSize); |
| |
| PairSharedBuffer composite; |
| composite.set(0, headers); |
| composite.set(1, payload); |
| |
| // Write checksum at created checksum-placeholder |
| if (includeChecksum) { |
| int writeIndex = headers.writerIndex(); |
| int metadataStartIndex = checksumReaderIndex + checksumSize; |
| uint32_t metadataChecksum = |
| computeChecksum(0, headers.data() + metadataStartIndex, (writeIndex - metadataStartIndex)); |
| uint32_t computedChecksum = |
| computeChecksum(metadataChecksum, payload.data(), payload.readableBytes()); |
| // set computed checksum |
| headers.setWriterIndex(checksumReaderIndex); |
| headers.writeUnsignedInt(computedChecksum); |
| headers.setWriterIndex(writeIndex); |
| } |
| |
| cmd.clear_send(); |
| return composite; |
| } |
| |
| SharedBuffer Commands::newConnect(const AuthenticationPtr& authentication, const std::string& logicalAddress, |
| bool connectingThroughProxy, const std::string& clientVersion, |
| Result& result) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::CONNECT); |
| CommandConnect* connect = cmd.mutable_connect(); |
| connect->set_client_version(clientVersion); |
| connect->set_auth_method_name(authentication->getAuthMethodName()); |
| connect->set_protocol_version(ProtocolVersion_MAX); |
| |
| FeatureFlags* flags = connect->mutable_feature_flags(); |
| flags->set_supports_auth_refresh(true); |
| if (connectingThroughProxy) { |
| Url logicalAddressUrl; |
| Url::parse(logicalAddress, logicalAddressUrl); |
| connect->set_proxy_to_broker_url(logicalAddressUrl.hostPort()); |
| } |
| |
| AuthenticationDataPtr authDataContent; |
| result = authentication->getAuthData(authDataContent); |
| if (result != ResultOk) { |
| return SharedBuffer{}; |
| } |
| |
| if (authDataContent->hasDataFromCommand()) { |
| connect->set_auth_data(authDataContent->getCommandData()); |
| } |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newAuthResponse(const AuthenticationPtr& authentication, Result& result) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::AUTH_RESPONSE); |
| CommandAuthResponse* authResponse = cmd.mutable_authresponse(); |
| authResponse->set_client_version(std::string("Pulsar-CPP-v") + PULSAR_VERSION_STR); |
| |
| AuthData* authData = authResponse->mutable_response(); |
| authData->set_auth_method_name(authentication->getAuthMethodName()); |
| |
| AuthenticationDataPtr authDataContent; |
| result = authentication->getAuthData(authDataContent); |
| if (result != ResultOk) { |
| return SharedBuffer{}; |
| } |
| |
| if (authDataContent->hasDataFromCommand()) { |
| authData->set_auth_data(authDataContent->getCommandData()); |
| } |
| |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newSubscribe(const std::string& topic, const std::string& subscription, |
| uint64_t consumerId, uint64_t requestId, CommandSubscribe_SubType subType, |
| const std::string& consumerName, SubscriptionMode subscriptionMode, |
| boost::optional<MessageId> startMessageId, bool readCompacted, |
| const std::map<std::string, std::string>& metadata, |
| const std::map<std::string, std::string>& subscriptionProperties, |
| const SchemaInfo& schemaInfo, |
| CommandSubscribe_InitialPosition subscriptionInitialPosition, |
| bool replicateSubscriptionState, KeySharedPolicy keySharedPolicy, |
| int priorityLevel) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::SUBSCRIBE); |
| CommandSubscribe* subscribe = cmd.mutable_subscribe(); |
| subscribe->set_topic(topic); |
| subscribe->set_subscription(subscription); |
| subscribe->set_subtype(static_cast<proto::CommandSubscribe_SubType>(subType)); |
| subscribe->set_consumer_id(consumerId); |
| subscribe->set_request_id(requestId); |
| subscribe->set_consumer_name(consumerName); |
| subscribe->set_durable(subscriptionMode == SubscriptionModeDurable); |
| subscribe->set_read_compacted(readCompacted); |
| subscribe->set_initialposition( |
| static_cast<proto::CommandSubscribe_InitialPosition>(subscriptionInitialPosition)); |
| subscribe->set_replicate_subscription_state(replicateSubscriptionState); |
| subscribe->set_priority_level(priorityLevel); |
| |
| if (isBuiltInSchema(schemaInfo.getSchemaType())) { |
| subscribe->set_allocated_schema(getSchema(schemaInfo)); |
| } |
| |
| if (startMessageId) { |
| MessageIdData& messageIdData = *subscribe->mutable_start_message_id(); |
| messageIdData.set_ledgerid(startMessageId.value().ledgerId()); |
| messageIdData.set_entryid(startMessageId.value().entryId()); |
| |
| if (startMessageId.value().batchIndex() != -1) { |
| messageIdData.set_batch_index(startMessageId.value().batchIndex()); |
| } |
| } |
| for (std::map<std::string, std::string>::const_iterator it = metadata.begin(); it != metadata.end(); |
| it++) { |
| proto::KeyValue* keyValue = proto::KeyValue().New(); |
| keyValue->set_key(it->first); |
| keyValue->set_value(it->second); |
| subscribe->mutable_metadata()->AddAllocated(keyValue); |
| } |
| |
| for (const auto& subscriptionProperty : subscriptionProperties) { |
| proto::KeyValue* keyValue = proto::KeyValue().New(); |
| keyValue->set_key(subscriptionProperty.first); |
| keyValue->set_value(subscriptionProperty.second); |
| subscribe->mutable_subscription_properties()->AddAllocated(keyValue); |
| } |
| |
| if (subType == CommandSubscribe_SubType_Key_Shared) { |
| KeySharedMeta& ksm = *subscribe->mutable_keysharedmeta(); |
| switch (keySharedPolicy.getKeySharedMode()) { |
| case pulsar::AUTO_SPLIT: |
| ksm.set_keysharedmode(proto::KeySharedMode::AUTO_SPLIT); |
| break; |
| case pulsar::STICKY: |
| ksm.set_keysharedmode(proto::KeySharedMode::STICKY); |
| for (StickyRange range : keySharedPolicy.getStickyRanges()) { |
| IntRange* intRange = IntRange().New(); |
| intRange->set_start(range.first); |
| intRange->set_end(range.second); |
| ksm.mutable_hashranges()->AddAllocated(intRange); |
| } |
| } |
| ksm.set_allowoutoforderdelivery(keySharedPolicy.isAllowOutOfOrderDelivery()); |
| } |
| |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newUnsubscribe(uint64_t consumerId, uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::UNSUBSCRIBE); |
| CommandUnsubscribe* unsubscribe = cmd.mutable_unsubscribe(); |
| unsubscribe->set_consumer_id(consumerId); |
| unsubscribe->set_request_id(requestId); |
| |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newProducer(const std::string& topic, uint64_t producerId, |
| const std::string& producerName, uint64_t requestId, |
| const std::map<std::string, std::string>& metadata, |
| const SchemaInfo& schemaInfo, uint64_t epoch, |
| bool userProvidedProducerName, bool encrypted, |
| ProducerAccessMode accessMode, boost::optional<uint64_t> topicEpoch, |
| const std::string& initialSubscriptionName) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::PRODUCER); |
| CommandProducer* producer = cmd.mutable_producer(); |
| producer->set_topic(topic); |
| producer->set_producer_id(producerId); |
| producer->set_request_id(requestId); |
| producer->set_epoch(epoch); |
| producer->set_user_provided_producer_name(userProvidedProducerName); |
| producer->set_encrypted(encrypted); |
| producer->set_producer_access_mode(static_cast<proto::ProducerAccessMode>(accessMode)); |
| if (topicEpoch) { |
| producer->set_topic_epoch(topicEpoch.value()); |
| } |
| if (!initialSubscriptionName.empty()) { |
| producer->set_initial_subscription_name(initialSubscriptionName); |
| } |
| |
| for (std::map<std::string, std::string>::const_iterator it = metadata.begin(); it != metadata.end(); |
| it++) { |
| proto::KeyValue* keyValue = proto::KeyValue().New(); |
| keyValue->set_key(it->first); |
| keyValue->set_value(it->second); |
| producer->mutable_metadata()->AddAllocated(keyValue); |
| } |
| |
| if (isBuiltInSchema(schemaInfo.getSchemaType())) { |
| producer->set_allocated_schema(getSchema(schemaInfo)); |
| } |
| |
| if (!producerName.empty()) { |
| producer->set_producer_name(producerName); |
| } |
| |
| return writeMessageWithSize(cmd); |
| } |
| |
| static void configureCommandAck(CommandAck* ack, uint64_t consumerId, int64_t ledgerId, int64_t entryId, |
| const BitSet& ackSet, CommandAck_AckType ackType) { |
| ack->set_consumer_id(consumerId); |
| ack->set_ack_type(static_cast<proto::CommandAck_AckType>(ackType)); |
| auto* msgId = ack->add_message_id(); |
| msgId->set_ledgerid(ledgerId); |
| msgId->set_entryid(entryId); |
| for (auto x : ackSet) { |
| msgId->add_ack_set(x); |
| } |
| } |
| |
| SharedBuffer Commands::newAck(uint64_t consumerId, int64_t ledgerId, int64_t entryId, const BitSet& ackSet, |
| CommandAck_AckType ackType) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::ACK); |
| configureCommandAck(cmd.mutable_ack(), consumerId, ledgerId, entryId, ackSet, ackType); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newAck(uint64_t consumerId, int64_t ledgerId, int64_t entryId, const BitSet& ackSet, |
| CommandAck_AckType ackType, CommandAck_ValidationError validationError) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::ACK); |
| CommandAck* ack = cmd.mutable_ack(); |
| ack->set_validation_error((proto::CommandAck_ValidationError)validationError); |
| configureCommandAck(ack, consumerId, ledgerId, entryId, ackSet, ackType); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newAck(uint64_t consumerId, int64_t ledgerId, int64_t entryId, const BitSet& ackSet, |
| CommandAck_AckType ackType, uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::ACK); |
| CommandAck* ack = cmd.mutable_ack(); |
| ack->set_request_id(requestId); |
| configureCommandAck(ack, consumerId, ledgerId, entryId, ackSet, ackType); |
| return writeMessageWithSize(cmd); |
| } |
| |
| static void configureCommandAck(CommandAck* ack, uint64_t consumerId, const std::set<MessageId>& msgIds) { |
| ack->set_consumer_id(consumerId); |
| ack->set_ack_type(proto::CommandAck_AckType_Individual); |
| for (const auto& msgId : msgIds) { |
| auto newMsgId = ack->add_message_id(); |
| newMsgId->set_ledgerid(msgId.ledgerId()); |
| newMsgId->set_entryid(msgId.entryId()); |
| for (auto x : Commands::getMessageIdImpl(msgId)->getBitSet()) { |
| newMsgId->add_ack_set(x); |
| } |
| } |
| } |
| |
| SharedBuffer Commands::newMultiMessageAck(uint64_t consumerId, const std::set<MessageId>& msgIds) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::ACK); |
| configureCommandAck(cmd.mutable_ack(), consumerId, msgIds); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newMultiMessageAck(uint64_t consumerId, const std::set<MessageId>& msgIds, |
| uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::ACK); |
| CommandAck* ack = cmd.mutable_ack(); |
| ack->set_request_id(requestId); |
| configureCommandAck(ack, consumerId, msgIds); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newFlow(uint64_t consumerId, uint32_t messagePermits) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::FLOW); |
| CommandFlow* flow = cmd.mutable_flow(); |
| flow->set_consumer_id(consumerId); |
| flow->set_messagepermits(messagePermits); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newCloseProducer(uint64_t producerId, uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::CLOSE_PRODUCER); |
| CommandCloseProducer* close = cmd.mutable_close_producer(); |
| close->set_producer_id(producerId); |
| close->set_request_id(requestId); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newCloseConsumer(uint64_t consumerId, uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::CLOSE_CONSUMER); |
| CommandCloseConsumer* close = cmd.mutable_close_consumer(); |
| close->set_consumer_id(consumerId); |
| close->set_request_id(requestId); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newPing() { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::PING); |
| cmd.mutable_ping(); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newPong() { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::PONG); |
| cmd.mutable_pong(); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newRedeliverUnacknowledgedMessages(uint64_t consumerId, |
| const std::set<MessageId>& messageIds) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::REDELIVER_UNACKNOWLEDGED_MESSAGES); |
| CommandRedeliverUnacknowledgedMessages* command = cmd.mutable_redeliverunacknowledgedmessages(); |
| command->set_consumer_id(consumerId); |
| for (const auto& msgId : messageIds) { |
| MessageIdData* msgIdData = command->add_message_ids(); |
| msgIdData->set_ledgerid(msgId.ledgerId()); |
| msgIdData->set_entryid(msgId.entryId()); |
| } |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newSeek(uint64_t consumerId, uint64_t requestId, const MessageId& messageId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::SEEK); |
| CommandSeek* commandSeek = cmd.mutable_seek(); |
| commandSeek->set_consumer_id(consumerId); |
| commandSeek->set_request_id(requestId); |
| |
| MessageIdData& messageIdData = *commandSeek->mutable_message_id(); |
| |
| auto chunkMsgId = std::dynamic_pointer_cast<ChunkMessageIdImpl>(messageId.impl_); |
| if (chunkMsgId) { |
| auto firstId = chunkMsgId->getFirstChunkMessageId(); |
| messageIdData.set_ledgerid(firstId->ledgerId_); |
| messageIdData.set_entryid(firstId->entryId_); |
| } else { |
| messageIdData.set_ledgerid(messageId.ledgerId()); |
| messageIdData.set_entryid(messageId.entryId()); |
| } |
| |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newSeek(uint64_t consumerId, uint64_t requestId, uint64_t timestamp) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::SEEK); |
| CommandSeek* commandSeek = cmd.mutable_seek(); |
| commandSeek->set_consumer_id(consumerId); |
| commandSeek->set_request_id(requestId); |
| commandSeek->set_message_publish_time(timestamp); |
| return writeMessageWithSize(cmd); |
| } |
| |
| SharedBuffer Commands::newGetLastMessageId(uint64_t consumerId, uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::GET_LAST_MESSAGE_ID); |
| |
| CommandGetLastMessageId* getLastMessageId = cmd.mutable_getlastmessageid(); |
| getLastMessageId->set_consumer_id(consumerId); |
| getLastMessageId->set_request_id(requestId); |
| const SharedBuffer buffer = writeMessageWithSize(cmd); |
| cmd.clear_getlastmessageid(); |
| return buffer; |
| } |
| |
| SharedBuffer Commands::newGetTopicsOfNamespace(const std::string& nsName, |
| CommandGetTopicsOfNamespace_Mode mode, uint64_t requestId) { |
| BaseCommand cmd; |
| cmd.set_type(BaseCommand::GET_TOPICS_OF_NAMESPACE); |
| CommandGetTopicsOfNamespace* getTopics = cmd.mutable_gettopicsofnamespace(); |
| getTopics->set_request_id(requestId); |
| getTopics->set_namespace_(nsName); |
| getTopics->set_mode(static_cast<proto::CommandGetTopicsOfNamespace_Mode>(mode)); |
| |
| const SharedBuffer buffer = writeMessageWithSize(cmd); |
| cmd.clear_gettopicsofnamespace(); |
| return buffer; |
| } |
| |
| std::string Commands::messageType(BaseCommand_Type type) { |
| switch (type) { |
| case BaseCommand::CONNECT: |
| return "CONNECT"; |
| break; |
| case BaseCommand::CONNECTED: |
| return "CONNECTED"; |
| break; |
| case BaseCommand::SUBSCRIBE: |
| return "SUBSCRIBE"; |
| break; |
| case BaseCommand::PRODUCER: |
| return "PRODUCER"; |
| break; |
| case BaseCommand::SEND: |
| return "SEND"; |
| break; |
| case BaseCommand::SEND_RECEIPT: |
| return "SEND_RECEIPT"; |
| break; |
| case BaseCommand::SEND_ERROR: |
| return "SEND_ERROR"; |
| break; |
| case BaseCommand::MESSAGE: |
| return "MESSAGE"; |
| break; |
| case BaseCommand::ACK: |
| return "ACK"; |
| break; |
| case BaseCommand::FLOW: |
| return "FLOW"; |
| break; |
| case BaseCommand::UNSUBSCRIBE: |
| return "UNSUBSCRIBE"; |
| break; |
| case BaseCommand::SUCCESS: |
| return "SUCCESS"; |
| break; |
| case BaseCommand::ERROR: |
| return "ERROR"; |
| break; |
| case BaseCommand::CLOSE_PRODUCER: |
| return "CLOSE_PRODUCER"; |
| break; |
| case BaseCommand::CLOSE_CONSUMER: |
| return "CLOSE_CONSUMER"; |
| break; |
| case BaseCommand::PRODUCER_SUCCESS: |
| return "PRODUCER_SUCCESS"; |
| break; |
| case BaseCommand::PING: |
| return "PING"; |
| break; |
| case BaseCommand::PONG: |
| return "PONG"; |
| break; |
| case BaseCommand::PARTITIONED_METADATA: |
| return "PARTITIONED_METADATA"; |
| break; |
| case BaseCommand::PARTITIONED_METADATA_RESPONSE: |
| return "PARTITIONED_METADATA_RESPONSE"; |
| break; |
| case BaseCommand::REDELIVER_UNACKNOWLEDGED_MESSAGES: |
| return "REDELIVER_UNACKNOWLEDGED_MESSAGES"; |
| break; |
| case BaseCommand::LOOKUP: |
| return "LOOKUP"; |
| break; |
| case BaseCommand::LOOKUP_RESPONSE: |
| return "LOOKUP_RESPONSE"; |
| break; |
| case BaseCommand::CONSUMER_STATS: |
| return "CONSUMER_STATS"; |
| break; |
| case BaseCommand::CONSUMER_STATS_RESPONSE: |
| return "CONSUMER_STATS_RESPONSE"; |
| break; |
| case BaseCommand::REACHED_END_OF_TOPIC: |
| return "REACHED_END_OF_TOPIC"; |
| break; |
| case BaseCommand::SEEK: |
| return "SEEK"; |
| break; |
| case BaseCommand::ACTIVE_CONSUMER_CHANGE: |
| return "ACTIVE_CONSUMER_CHANGE"; |
| break; |
| case BaseCommand::GET_LAST_MESSAGE_ID: |
| return "GET_LAST_MESSAGE_ID"; |
| break; |
| case BaseCommand::GET_LAST_MESSAGE_ID_RESPONSE: |
| return "GET_LAST_MESSAGE_ID_RESPONSE"; |
| break; |
| case BaseCommand::GET_TOPICS_OF_NAMESPACE: |
| return "GET_TOPICS_OF_NAMESPACE"; |
| break; |
| case BaseCommand::GET_TOPICS_OF_NAMESPACE_RESPONSE: |
| return "GET_TOPICS_OF_NAMESPACE_RESPONSE"; |
| break; |
| case BaseCommand::GET_SCHEMA: |
| return "GET_SCHEMA"; |
| break; |
| case BaseCommand::GET_SCHEMA_RESPONSE: |
| return "GET_SCHEMA_RESPONSE"; |
| break; |
| case BaseCommand::AUTH_CHALLENGE: |
| return "AUTH_CHALLENGE"; |
| break; |
| case BaseCommand::AUTH_RESPONSE: |
| return "AUTH_RESPONSE"; |
| break; |
| case BaseCommand::ACK_RESPONSE: |
| return "ACK_RESPONSE"; |
| break; |
| case BaseCommand::GET_OR_CREATE_SCHEMA: |
| return "GET_OR_CREATE_SCHEMA"; |
| case BaseCommand::GET_OR_CREATE_SCHEMA_RESPONSE: |
| return "GET_OR_CREATE_SCHEMA_RESPONSE"; |
| case BaseCommand::NEW_TXN: |
| return "NEW_TXN"; |
| break; |
| case BaseCommand::NEW_TXN_RESPONSE: |
| return "NEW_TXN_RESPONSE"; |
| break; |
| case BaseCommand::ADD_PARTITION_TO_TXN: |
| return "ADD_PARTITION_TO_TXN"; |
| break; |
| case BaseCommand::ADD_PARTITION_TO_TXN_RESPONSE: |
| return "ADD_PARTITION_TO_TXN_RESPONSE"; |
| break; |
| case BaseCommand::ADD_SUBSCRIPTION_TO_TXN: |
| return "ADD_SUBSCRIPTION_TO_TXN"; |
| break; |
| case BaseCommand::ADD_SUBSCRIPTION_TO_TXN_RESPONSE: |
| return "ADD_SUBSCRIPTION_TO_TXN_RESPONSE"; |
| break; |
| case BaseCommand::END_TXN: |
| return "END_TXN"; |
| break; |
| case BaseCommand::END_TXN_RESPONSE: |
| return "END_TXN_RESPONSE"; |
| break; |
| case BaseCommand::END_TXN_ON_PARTITION: |
| return "END_TXN_ON_PARTITION"; |
| break; |
| case BaseCommand::END_TXN_ON_PARTITION_RESPONSE: |
| return "END_TXN_ON_PARTITION_RESPONSE"; |
| break; |
| case BaseCommand::END_TXN_ON_SUBSCRIPTION: |
| return "END_TXN_ON_SUBSCRIPTION"; |
| break; |
| case BaseCommand::END_TXN_ON_SUBSCRIPTION_RESPONSE: |
| return "END_TXN_ON_SUBSCRIPTION_RESPONSE"; |
| break; |
| case BaseCommand::TC_CLIENT_CONNECT_REQUEST: |
| return "TC_CLIENT_CONNECT_REQUEST"; |
| case BaseCommand::TC_CLIENT_CONNECT_RESPONSE: |
| return "TC_CLIENT_CONNECT_RESPONSE"; |
| break; |
| case BaseCommand::WATCH_TOPIC_LIST: |
| return "WATCH_TOPIC_LIST"; |
| break; |
| case BaseCommand::WATCH_TOPIC_LIST_SUCCESS: |
| return "WATCH_TOPIC_LIST_SUCCESS"; |
| break; |
| case BaseCommand::WATCH_TOPIC_UPDATE: |
| return "WATCH_TOPIC_UPDATE"; |
| break; |
| case BaseCommand::WATCH_TOPIC_LIST_CLOSE: |
| return "WATCH_TOPIC_LIST_CLOSE"; |
| break; |
| }; |
| BOOST_THROW_EXCEPTION(std::logic_error("Invalid BaseCommand enumeration value")); |
| } |
| |
| void Commands::initBatchMessageMetadata(const Message& msg, pulsar::proto::MessageMetadata& batchMetadata) { |
| // metadata has already been set in ProducerImpl::setMessageMetadata |
| const proto::MessageMetadata& metadata = msg.impl_->metadata; |
| |
| // required fields |
| batchMetadata.set_producer_name(metadata.producer_name()); |
| batchMetadata.set_sequence_id(metadata.sequence_id()); |
| batchMetadata.set_publish_time(metadata.publish_time()); |
| |
| // optional fields |
| if (metadata.has_partition_key()) { |
| batchMetadata.set_partition_key(metadata.partition_key()); |
| } |
| if (metadata.has_ordering_key()) { |
| batchMetadata.set_ordering_key(metadata.ordering_key()); |
| } |
| if (metadata.has_replicated_from()) { |
| batchMetadata.set_replicated_from(metadata.replicated_from()); |
| } |
| if (metadata.replicate_to_size() > 0) { |
| for (int i = 0; i < metadata.replicate_to_size(); i++) { |
| batchMetadata.add_replicate_to(metadata.replicate_to(i)); |
| } |
| } |
| if (metadata.has_schema_version()) { |
| batchMetadata.set_schema_version(metadata.schema_version()); |
| } |
| } |
| |
| uint64_t Commands::serializeSingleMessageInBatchWithPayload(const Message& msg, SharedBuffer& batchPayLoad, |
| unsigned long maxMessageSizeInBytes) { |
| const auto& msgMetadata = msg.impl_->metadata; |
| SingleMessageMetadata metadata; |
| if (msgMetadata.has_partition_key()) { |
| metadata.set_partition_key(msgMetadata.partition_key()); |
| } |
| if (msgMetadata.has_ordering_key()) { |
| metadata.set_ordering_key(msgMetadata.ordering_key()); |
| } |
| |
| metadata.mutable_properties()->Reserve(msgMetadata.properties_size()); |
| for (int i = 0; i < msgMetadata.properties_size(); i++) { |
| auto keyValue = proto::KeyValue().New(); |
| *keyValue = msgMetadata.properties(i); |
| metadata.mutable_properties()->AddAllocated(keyValue); |
| } |
| |
| if (msgMetadata.has_event_time()) { |
| metadata.set_event_time(msgMetadata.event_time()); |
| } |
| |
| if (msgMetadata.has_sequence_id()) { |
| metadata.set_sequence_id(msgMetadata.sequence_id()); |
| } |
| |
| // Format of batch message |
| // Each Message = [METADATA_SIZE][METADATA] [PAYLOAD] |
| |
| int payloadSize = msg.impl_->payload.readableBytes(); |
| metadata.set_payload_size(payloadSize); |
| |
| int msgMetadataSize = metadata.ByteSize(); |
| |
| unsigned long requiredSpace = sizeof(uint32_t) + msgMetadataSize + payloadSize; |
| if (batchPayLoad.writableBytes() <= sizeof(uint32_t) + msgMetadataSize + payloadSize) { |
| LOG_DEBUG("remaining size of batchPayLoad buffer [" |
| << batchPayLoad.writableBytes() << "] can't accomodate new payload [" << requiredSpace |
| << "] - expanding the batchPayload buffer"); |
| uint32_t new_size = |
| std::min(batchPayLoad.readableBytes() * 2, static_cast<uint32_t>(maxMessageSizeInBytes)); |
| new_size = std::max(new_size, batchPayLoad.readableBytes() + static_cast<uint32_t>(requiredSpace)); |
| SharedBuffer buffer = SharedBuffer::allocate(new_size); |
| // Adding batch created so far |
| buffer.write(batchPayLoad.data(), batchPayLoad.readableBytes()); |
| batchPayLoad = buffer; |
| } |
| // Adding the new message |
| batchPayLoad.writeUnsignedInt(msgMetadataSize); |
| metadata.SerializeToArray(batchPayLoad.mutableData(), msgMetadataSize); |
| batchPayLoad.bytesWritten(msgMetadataSize); |
| batchPayLoad.write(msg.impl_->payload.data(), payloadSize); |
| |
| return msgMetadata.sequence_id(); |
| } |
| |
| Message Commands::deSerializeSingleMessageInBatch(Message& batchedMessage, int32_t batchIndex, |
| int32_t batchSize, const BatchMessageAckerPtr& acker) { |
| SharedBuffer& uncompressedPayload = batchedMessage.impl_->payload; |
| |
| // Format of batch message |
| // Each Message = [METADATA_SIZE][METADATA] [PAYLOAD] |
| |
| const int& singleMetaSize = uncompressedPayload.readUnsignedInt(); |
| SingleMessageMetadata metadata; |
| metadata.ParseFromArray(uncompressedPayload.data(), singleMetaSize); |
| uncompressedPayload.consume(singleMetaSize); |
| |
| const int& payloadSize = metadata.payload_size(); |
| |
| // Get a slice of size payloadSize from offset readIndex_ |
| SharedBuffer payload = uncompressedPayload.slice(0, payloadSize); |
| uncompressedPayload.consume(payloadSize); |
| |
| 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_->metadata, payload, metadata, |
| batchedMessage.impl_->topicName_); |
| singleMessage.impl_->cnx_ = batchedMessage.impl_->cnx_; |
| |
| return singleMessage; |
| } |
| |
| MessageIdImplPtr Commands::getMessageIdImpl(const MessageId& messageId) { return messageId.impl_; } |
| |
| bool Commands::peerSupportsGetLastMessageId(int32_t peerVersion) { return peerVersion >= proto::v12; } |
| |
| bool Commands::peerSupportsActiveConsumerListener(int32_t peerVersion) { return peerVersion >= proto::v12; } |
| |
| bool Commands::peerSupportsMultiMessageAcknowledgement(int32_t peerVersion) { |
| return peerVersion >= proto::v12; |
| } |
| |
| bool Commands::peerSupportsJsonSchemaAvroFormat(int32_t peerVersion) { return peerVersion >= proto::v13; } |
| |
| bool Commands::peerSupportsGetOrCreateSchema(int32_t peerVersion) { return peerVersion >= proto::v15; } |
| |
| } // namespace pulsar |
| /* namespace pulsar */ |
