From 7bfa7212ccc8a615785c936d6eb9111879ec7960 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Fri, 2 Sep 2022 17:12:09 +0800 Subject: [PATCH 1/6] [feat][cpp] Consumer support batch receive messages. --- pulsar-client-cpp/.gitignore | 1 + .../include/pulsar/BatchReceivePolicy.h | 87 ++++++++ pulsar-client-cpp/include/pulsar/Consumer.h | 26 +++ .../include/pulsar/ConsumerConfiguration.h | 17 ++ pulsar-client-cpp/include/pulsar/Messages.h | 51 +++++ pulsar-client-cpp/lib/BatchReceivePolicy.cc | 57 ++++++ .../lib/BatchReceivePolicyImpl.h | 29 +++ pulsar-client-cpp/lib/ClientImpl.h | 5 +- pulsar-client-cpp/lib/Consumer.cc | 18 ++ .../lib/ConsumerConfiguration.cc | 9 + .../lib/ConsumerConfigurationImpl.h | 1 + pulsar-client-cpp/lib/ConsumerImpl.cc | 179 +++++++++-------- pulsar-client-cpp/lib/ConsumerImpl.h | 15 +- pulsar-client-cpp/lib/ConsumerImplBase.cc | 144 ++++++++++++++ pulsar-client-cpp/lib/ConsumerImplBase.h | 44 +++- pulsar-client-cpp/lib/HandlerBase.h | 1 + pulsar-client-cpp/lib/Messages.cc | 34 ++++ pulsar-client-cpp/lib/MessagesImpl.cc | 62 ++++++ pulsar-client-cpp/lib/MessagesImpl.h | 48 +++++ .../lib/MultiTopicsConsumerImpl.cc | 146 ++++++++++---- .../lib/MultiTopicsConsumerImpl.h | 22 +- pulsar-client-cpp/tests/BasicEndToEndTest.cc | 188 ++++++++++++++++++ .../tests/BatchReceivePolicyTest.cc | 45 +++++ .../tests/ConsumerConfigurationTest.cc | 8 + pulsar-client-cpp/tests/MessagesTest.cc | 72 +++++++ 25 files changed, 1160 insertions(+), 149 deletions(-) create mode 100644 pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h create mode 100644 pulsar-client-cpp/include/pulsar/Messages.h create mode 100644 pulsar-client-cpp/lib/BatchReceivePolicy.cc create mode 100644 pulsar-client-cpp/lib/BatchReceivePolicyImpl.h create mode 100644 pulsar-client-cpp/lib/ConsumerImplBase.cc create mode 100644 pulsar-client-cpp/lib/Messages.cc create mode 100644 pulsar-client-cpp/lib/MessagesImpl.cc create mode 100644 pulsar-client-cpp/lib/MessagesImpl.h create mode 100644 pulsar-client-cpp/tests/BatchReceivePolicyTest.cc create mode 100644 pulsar-client-cpp/tests/MessagesTest.cc diff --git a/pulsar-client-cpp/.gitignore b/pulsar-client-cpp/.gitignore index 8c8c065e61935..9111dd42202f5 100644 --- a/pulsar-client-cpp/.gitignore +++ b/pulsar-client-cpp/.gitignore @@ -77,6 +77,7 @@ Makefile cmake_install.cmake CMakeFiles CMakeCache.txt +build/ pulsar-dist install_manifest.txt diff --git a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h new file mode 100644 index 0000000000000..ca9c4046b40fc --- /dev/null +++ b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h @@ -0,0 +1,87 @@ +/** + * 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. + */ +#ifndef BATCH_RECEIVE_POLICY_HPP_ +#define BATCH_RECEIVE_POLICY_HPP_ + +#include +#include + +namespace pulsar { + +struct BatchReceivePolicyImpl; + +/** + * Configuration for message batch receive {@link Consumer#batchReceive()} {@link + * Consumer#batchReceiveAsync()}. + * + *

Batch receive policy can limit the number and bytes of messages in a single batch, and can specify a + * timeout for waiting for enough messages for this batch. + * + *

This batch receive will be completed as long as any one of the + * conditions(has enough number of messages, has enough of size of messages, wait timeout) is met. + * + *

Examples: + * 1.If set maxNumMessages = 10, maxSizeOfMessages = 1MB and without timeout, it + * means {@link Consumer#batchReceive()} will always wait until there is enough messages. + * 2.If set maxNumberOfMessages = 0, maxNumBytes = 0 and timeout = 100ms, it + * means {@link Consumer#batchReceive()} will waiting for 100ms whether or not there is enough messages. + * + *

Note: + * Must specify messages limitation(maxNumMessages, maxNumBytes) or wait timeout. + * Otherwise, {@link Messages} ingest {@link Message} will never end. + * + * @since 2.4.1 + */ +class PULSAR_PUBLIC BatchReceivePolicy { + public: + BatchReceivePolicy(); + + /** + * + * @param maxNumMessage Max num message, if less than 0, it means no limit. + * @param maxNumBytes Max num bytes, if less than 0, it means no limit. + * @param timeoutMs If less than 0, it means no limit. + */ + BatchReceivePolicy(int maxNumMessage, long maxNumBytes, long timeoutMs); + + /** + * Get max time out ms. + * + * @return + */ + long getTimeoutMs() const; + + /** + * Get max num messages. + * @return + */ + int getMaxNumMessages() const; + + /** + * Get max num bytes. + * @return + */ + long getMaxNumBytes() const; + + private: + std::shared_ptr impl_; +}; +} // namespace pulsar + +#endif /* BATCH_RECEIVE_POLICY_HPP_ */ diff --git a/pulsar-client-cpp/include/pulsar/Consumer.h b/pulsar-client-cpp/include/pulsar/Consumer.h index 6c0ab27b06c75..907d26af93f12 100644 --- a/pulsar-client-cpp/include/pulsar/Consumer.h +++ b/pulsar-client-cpp/include/pulsar/Consumer.h @@ -23,6 +23,7 @@ #include #include #include +#include namespace pulsar { class PulsarWrapper; @@ -113,6 +114,31 @@ class PULSAR_PUBLIC Consumer { */ void receiveAsync(ReceiveCallback callback); + /** + * Batch receiving messages. + * + *

This calls blocks until has enough messages or wait timeout, more details to see {@link + * BatchReceivePolicy}. + * + * @param msgs a non-const reference where the received messages will be copied + * @return ResultOk when a message is received + * @return ResultInvalidConfiguration if a message listener had been set in the configuration + */ + Result batchReceive(Messages& msgs); + + /** + * Async Batch receiving messages. + *

+ * Retrieves a message when it will be available and completes callback with received message. + *

+ *

+ * batchReceiveAsync() should be called subsequently once callback gets completed with received message. + * Else it creates backlog of receive requests in the application. + *

+ * @param BatchReceiveCallback will be completed when messages is available + */ + void batchReceiveAsync(BatchReceiveCallback callback); + /** * Acknowledge the reception of a single message. * diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index b326ca8fb3151..c6fff393754ab 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -31,6 +31,8 @@ #include #include #include +#include +#include "BatchReceivePolicy.h" namespace pulsar { @@ -40,6 +42,7 @@ class PulsarWrapper; /// Callback definition for non-data operation typedef std::function ResultCallback; typedef std::function ReceiveCallback; +typedef std::function BatchReceiveCallback; typedef std::function GetLastMessageIdCallback; /// Callback definition for MessageListener @@ -378,6 +381,20 @@ class PULSAR_PUBLIC ConsumerConfiguration { */ InitialPosition getSubscriptionInitialPosition() const; + /** + * Set batch receive policy. + * + * @param batchReceivePolicy the default is xxx + */ + void setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy); + + /** + * Get batch receive policy. + * + * @return batch receive policy + */ + const BatchReceivePolicy& getBatchReceivePolicy() const; + /** * Set whether the subscription status should be replicated. * The default value is `false`. diff --git a/pulsar-client-cpp/include/pulsar/Messages.h b/pulsar-client-cpp/include/pulsar/Messages.h new file mode 100644 index 0000000000000..f0157f057ac1d --- /dev/null +++ b/pulsar-client-cpp/include/pulsar/Messages.h @@ -0,0 +1,51 @@ +/** + * 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. + */ +#ifndef MESSAGES_HPP_ +#define MESSAGES_HPP_ + +#include +#include +#include + +namespace pulsar { + +class Message; +class MessagesImpl; + +class PULSAR_PUBLIC Messages { + public: + Messages(); + + /** + * Get message list. + * + * @return message list. + */ + std::vector getMessageList() const; + + private: + typedef std::shared_ptr MessagesImplPtr; + MessagesImplPtr impl_; + Messages(MessagesImplPtr msgsPtr); + friend class ConsumerImpl; + friend class MultiTopicsConsumerImpl; +}; +} // namespace pulsar + +#endif /* MESSAGES_HPP_ */ diff --git a/pulsar-client-cpp/lib/BatchReceivePolicy.cc b/pulsar-client-cpp/lib/BatchReceivePolicy.cc new file mode 100644 index 0000000000000..08aa3687b58a6 --- /dev/null +++ b/pulsar-client-cpp/lib/BatchReceivePolicy.cc @@ -0,0 +1,57 @@ +/** + * 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 +#include "BatchReceivePolicyImpl.h" +#include "LogUtils.h" + +using namespace pulsar; + +namespace pulsar { + +DECLARE_LOG_OBJECT() + +BatchReceivePolicy::BatchReceivePolicy() : BatchReceivePolicy(-1, 10 * 1024 * 1024, 100) {} + +BatchReceivePolicy::BatchReceivePolicy(int maxNumMessage, long maxNumBytes, long timeoutMs) + : impl_(std::make_shared()) { + if (maxNumMessage <= 0 && maxNumBytes <= 0 && timeoutMs <= 0) { + throw std::invalid_argument( + "At least one of maxNumMessages, maxNumBytes and timeoutMs must be specified."); + } + if (maxNumMessage <= 0 && maxNumBytes <= 0) { + impl_->maxNumMessage = -1; + impl_->maxNumBytes = 10 * 1024 * 1024; + LOG_WARN( + "BatchReceivePolicy maxNumMessages and maxNumBytes is less than 0. Reset to default: " + "maxNumMessage(-1), maxNumBytes(10 * 1024 * 10)"); + } else { + impl_->maxNumMessage = maxNumMessage; + impl_->maxNumBytes = maxNumBytes; + } + impl_->timeoutMs = timeoutMs; +} + +long BatchReceivePolicy::getTimeoutMs() const { return impl_->timeoutMs; } + +int BatchReceivePolicy::getMaxNumMessages() const { return impl_->maxNumMessage; } + +long BatchReceivePolicy::getMaxNumBytes() const { return impl_->maxNumBytes; } + +} // namespace pulsar diff --git a/pulsar-client-cpp/lib/BatchReceivePolicyImpl.h b/pulsar-client-cpp/lib/BatchReceivePolicyImpl.h new file mode 100644 index 0000000000000..693dac01745ca --- /dev/null +++ b/pulsar-client-cpp/lib/BatchReceivePolicyImpl.h @@ -0,0 +1,29 @@ +/** + * 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. + */ +using namespace pulsar; + +namespace pulsar { + +struct BatchReceivePolicyImpl { + int maxNumMessage; + long maxNumBytes; + long timeoutMs; +}; + +} // namespace pulsar diff --git a/pulsar-client-cpp/lib/ClientImpl.h b/pulsar-client-cpp/lib/ClientImpl.h index 466461ae71ea0..e8e7708279ae0 100644 --- a/pulsar-client-cpp/lib/ClientImpl.h +++ b/pulsar-client-cpp/lib/ClientImpl.h @@ -28,14 +28,12 @@ #include #include #include "ProducerImplBase.h" -#include "ConsumerImplBase.h" #include #include #include "ServiceNameResolver.h" namespace pulsar { -class ClientImpl; class PulsarFriend; typedef std::shared_ptr ClientImplPtr; typedef std::weak_ptr ClientImplWeakPtr; @@ -44,6 +42,9 @@ class ReaderImpl; typedef std::shared_ptr ReaderImplPtr; typedef std::weak_ptr ReaderImplWeakPtr; +class ConsumerImplBase; +typedef std::weak_ptr ConsumerImplBaseWeakPtr; + std::string generateRandomName(); class ClientImpl : public std::enable_shared_from_this { diff --git a/pulsar-client-cpp/lib/Consumer.cc b/pulsar-client-cpp/lib/Consumer.cc index 5d1636291286b..13fb9f4a922e6 100644 --- a/pulsar-client-cpp/lib/Consumer.cc +++ b/pulsar-client-cpp/lib/Consumer.cc @@ -82,6 +82,24 @@ void Consumer::receiveAsync(ReceiveCallback callback) { impl_->receiveAsync(callback); } +Result Consumer::batchReceive(Messages& msgs) { + if (!impl_) { + return ResultConsumerNotInitialized; + } + Promise promise; + impl_->batchReceiveAsync(WaitForCallbackValue(promise)); + return promise.getFuture().get(msgs); +} + +void Consumer::batchReceiveAsync(BatchReceiveCallback callback) { + if (!impl_) { + Messages msgs; + callback(ResultConsumerNotInitialized, msgs); + return; + } + impl_->batchReceiveAsync(callback); +} + Result Consumer::acknowledge(const Message& message) { return acknowledge(message.getMessageId()); } Result Consumer::acknowledge(const MessageId& messageId) { diff --git a/pulsar-client-cpp/lib/ConsumerConfiguration.cc b/pulsar-client-cpp/lib/ConsumerConfiguration.cc index 2b58835cdbea3..103dea1827ad6 100644 --- a/pulsar-client-cpp/lib/ConsumerConfiguration.cc +++ b/pulsar-client-cpp/lib/ConsumerConfiguration.cc @@ -19,6 +19,7 @@ #include #include +#include namespace pulsar { @@ -260,4 +261,12 @@ bool ConsumerConfiguration::isAutoAckOldestChunkedMessageOnQueueFull() const { return impl_->autoAckOldestChunkedMessageOnQueueFull; } +void ConsumerConfiguration::setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy) { + impl_->batchReceivePolicy = batchReceivePolicy; +} + +const BatchReceivePolicy& ConsumerConfiguration::getBatchReceivePolicy() const { + return impl_->batchReceivePolicy; +} + } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h index 1c13f729b55e0..f7f38fcbbb483 100644 --- a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h +++ b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h @@ -45,6 +45,7 @@ struct ConsumerConfigurationImpl { ConsumerCryptoFailureAction cryptoFailureAction{ConsumerCryptoFailureAction::FAIL}; bool readCompacted{false}; InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest}; + BatchReceivePolicy batchReceivePolicy{}; int patternAutoDiscoveryPeriod{60}; bool replicateSubscriptionStateEnabled{false}; std::map properties; diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 79c20d84649b3..c68fec7fb0077 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -18,6 +18,7 @@ */ #include "ConsumerImpl.h" #include "MessageImpl.h" +#include "MessagesImpl.h" #include "Commands.h" #include "LogUtils.h" #include "TimeUtils.h" @@ -42,7 +43,8 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, bool hasParent /* = false by default */, const ConsumerTopicType consumerTopicType /* = NonPartitioned by default */, Commands::SubscriptionMode subscriptionMode, Optional startMessageId) - : HandlerBase(client, topic, Backoff(milliseconds(100), seconds(60), milliseconds(0))), + : ConsumerImplBase(client, topic, Backoff(milliseconds(100), seconds(60), milliseconds(0)), conf, + listenerExecutor ? listenerExecutor : client->getListenerExecutorProvider()->get()), waitingForZeroQueueSizeMessage(false), config_(conf), subscription_(subscriptionName), @@ -83,13 +85,6 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, unAckedMessageTrackerPtr_.reset(new UnAckedMessageTrackerDisabled()); } - // Initialize listener executor. - if (listenerExecutor) { - listenerExecutor_ = listenerExecutor; - } else { - listenerExecutor_ = client->getListenerExecutorProvider()->get(); - } - // Setup stats reporter. unsigned int statsIntervalInSeconds = client->getClientConfig().getStatsIntervalInSeconds(); if (statsIntervalInSeconds) { @@ -143,12 +138,12 @@ const std::string& ConsumerImpl::getTopic() const { return topic_; } void ConsumerImpl::start() { HandlerBase::start(); - // Initialize ackGroupingTrackerPtr_ here because the shared_from_this() was not initialized until the + // Initialize ackGroupingTrackerPtr_ here because the get_shared_this_ptr() was not initialized until the // constructor completed. if (TopicName::get(topic_)->isPersistent()) { if (config_.getAckGroupingTimeMs() > 0) { ackGroupingTrackerPtr_.reset(new AckGroupingTrackerEnabled( - client_.lock(), shared_from_this(), consumerId_, config_.getAckGroupingTimeMs(), + client_.lock(), get_shared_this_ptr(), consumerId_, config_.getAckGroupingTimeMs(), config_.getAckGroupingMaxSize())); } else { ackGroupingTrackerPtr_.reset(new AckGroupingTrackerDisabled(*this, consumerId_)); @@ -167,7 +162,7 @@ void ConsumerImpl::connectionOpened(const ClientConnectionPtr& cnx) { // Register consumer so that we can handle other incomming commands (e.g. ACTIVE_CONSUMER_CHANGE) after // sending the subscribe request. - cnx->registerConsumer(consumerId_, shared_from_this()); + cnx->registerConsumer(consumerId_, get_shared_this_ptr()); Lock lockForMessageId(mutexForMessageId_); Optional firstMessageInQueue = clearReceiveQueue(); @@ -190,13 +185,13 @@ void ConsumerImpl::connectionOpened(const ClientConnectionPtr& cnx) { config_.getSchema(), getInitialPosition(), config_.isReplicateSubscriptionStateEnabled(), config_.getKeySharedPolicy(), config_.getPriorityLevel()); cnx->sendRequestWithId(cmd, requestId) - .addListener( - std::bind(&ConsumerImpl::handleCreateConsumer, shared_from_this(), cnx, std::placeholders::_1)); + .addListener(std::bind(&ConsumerImpl::handleCreateConsumer, get_shared_this_ptr(), cnx, + std::placeholders::_1)); } void ConsumerImpl::connectionFailed(Result result) { // Keep a reference to ensure object is kept alive - ConsumerImplPtr ptr = shared_from_this(); + auto ptr = get_shared_this_ptr(); if (consumerCreatedPromise_.setFailed(result)) { state_ = Failed; @@ -239,7 +234,7 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r sendFlowPermitsToBroker(cnx, 1); } } - consumerCreatedPromise_.setValue(shared_from_this()); + consumerCreatedPromise_.setValue(get_shared_this_ptr()); } else { if (result == ResultTimeout) { // Creating the consumer has timed out. We need to ensure the broker closes the consumer @@ -252,12 +247,12 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r if (consumerCreatedPromise_.isComplete()) { // Consumer had already been initially created, we need to retry connecting in any case LOG_WARN(getName() << "Failed to reconnect consumer: " << strResult(result)); - scheduleReconnection(shared_from_this()); + scheduleReconnection(get_shared_this_ptr()); } else { // Consumer was not yet created, retry to connect to broker if it's possible if (isRetriableError(result) && (creationTimestamp_ + operationTimeut_ < TimeUtils::now())) { LOG_WARN(getName() << "Temporary error in creating consumer : " << strResult(result)); - scheduleReconnection(shared_from_this()); + scheduleReconnection(get_shared_this_ptr()); } else { LOG_ERROR(getName() << "Failed to create consumer: " << strResult(result)); consumerCreatedPromise_.setFailed(result); @@ -287,7 +282,7 @@ void ConsumerImpl::unsubscribeAsync(ResultCallback callback) { int requestId = client->newRequestId(); SharedBuffer cmd = Commands::newUnsubscribe(consumerId_, requestId); cnx->sendRequestWithId(cmd, requestId) - .addListener(std::bind(&ConsumerImpl::handleUnsubscribe, shared_from_this(), + .addListener(std::bind(&ConsumerImpl::handleUnsubscribe, get_shared_this_ptr(), std::placeholders::_1, callback)); } else { Result result = ResultNotConnected; @@ -446,33 +441,7 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: Lock lock(mutex_); numOfMessageReceived = receiveIndividualMessagesFromBatch(cnx, m, msg.redelivery_count()); } else { - Lock lock(pendingReceiveMutex_); - // if asyncReceive is waiting then notify callback without adding to incomingMessages queue - bool asyncReceivedWaiting = !pendingReceives_.empty(); - ReceiveCallback callback; - if (asyncReceivedWaiting) { - callback = pendingReceives_.front(); - pendingReceives_.pop(); - } - lock.unlock(); - - if (asyncReceivedWaiting) { - listenerExecutor_->postWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback, - shared_from_this(), ResultOk, m, callback)); - return; - } - - // config_.getReceiverQueueSize() != 0 or waiting For ZeroQueueSize Message` - if (config_.getReceiverQueueSize() != 0 || - (config_.getReceiverQueueSize() == 0 && messageListener_)) { - incomingMessages_.push(m); - } else { - Lock lock(mutex_); - if (waitingForZeroQueueSizeMessage) { - lock.unlock(); - incomingMessages_.push(m); - } - } + executeNotifyCallback(m); } if (messageListener_) { @@ -481,7 +450,7 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: } // Trigger message listener callback in a separate thread while (numOfMessageReceived--) { - listenerExecutor_->postWork(std::bind(&ConsumerImpl::internalListener, shared_from_this())); + listenerExecutor_->postWork(std::bind(&ConsumerImpl::internalListener, get_shared_this_ptr())); } } } @@ -489,16 +458,16 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: void ConsumerImpl::activeConsumerChanged(bool isActive) { if (eventListener_) { listenerExecutor_->postWork( - std::bind(&ConsumerImpl::internalConsumerChangeListener, shared_from_this(), isActive)); + std::bind(&ConsumerImpl::internalConsumerChangeListener, get_shared_this_ptr(), isActive)); } } void ConsumerImpl::internalConsumerChangeListener(bool isActive) { try { if (isActive) { - eventListener_->becameActive(Consumer(shared_from_this()), partitionIndex_); + eventListener_->becameActive(Consumer(get_shared_this_ptr()), partitionIndex_); } else { - eventListener_->becameInactive(Consumer(shared_from_this()), partitionIndex_); + eventListener_->becameInactive(Consumer(get_shared_this_ptr()), partitionIndex_); } } catch (const std::exception& e) { LOG_ERROR(getName() << "Exception thrown from event listener " << e.what()); @@ -512,11 +481,60 @@ void ConsumerImpl::failPendingReceiveCallback() { ReceiveCallback callback = pendingReceives_.front(); pendingReceives_.pop(); listenerExecutor_->postWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback, - shared_from_this(), ResultAlreadyClosed, msg, callback)); + get_shared_this_ptr(), ResultAlreadyClosed, msg, callback)); } lock.unlock(); } +void ConsumerImpl::executeNotifyCallback(Message& msg) { + Lock lock(pendingReceiveMutex_); + // if asyncReceive is waiting then notify callback without adding to incomingMessages queue + bool asyncReceivedWaiting = !pendingReceives_.empty(); + ReceiveCallback callback; + if (asyncReceivedWaiting) { + callback = pendingReceives_.front(); + pendingReceives_.pop(); + } + lock.unlock(); + + // has pending receive, direct callback. + if (asyncReceivedWaiting) { + listenerExecutor_->postWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback, + get_shared_this_ptr(), ResultOk, msg, callback)); + return; + } + + // try to add incoming messages. + // config_.getReceiverQueueSize() != 0 or waiting For ZeroQueueSize Message` + if (messageListener_ || config_.getReceiverQueueSize() != 0 || waitingForZeroQueueSizeMessage) { + incomingMessages_.push(msg); + incomingMessagesSize_.fetch_add(msg.getLength()); + } + + // try trigger pending batch messages + if (hasEnoughMessagesForBatchReceive()) { + ConsumerImplBase::notifyBatchPendingReceivedCallback(); + } +} + +void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) { + auto messages = std::make_shared(batchReceivePolicy_.getMaxNumMessages(), + batchReceivePolicy_.getMaxNumBytes()); + Message peekMsg; + while (incomingMessages_.peek(peekMsg) && messages->canAdd(peekMsg)) { + // decreaseIncomingMessageSize + Message msg; + incomingMessages_.pop(msg); + messageProcessed(msg); + messages->add(msg); + } + auto self = get_shared_this_ptr(); + listenerExecutor_->postWork([callback, messages, self]() { + Messages msgs(messages); + callback(ResultOk, msgs); + }); +} + void ConsumerImpl::notifyPendingReceivedCallback(Result result, Message& msg, const ReceiveCallback& callback) { if (result == ResultOk && config_.getReceiverQueueSize() != 0) { @@ -560,19 +578,7 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection } } - // - Lock lock(pendingReceiveMutex_); - if (!pendingReceives_.empty()) { - ReceiveCallback callback = pendingReceives_.front(); - pendingReceives_.pop(); - lock.unlock(); - listenerExecutor_->postWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback, - shared_from_this(), ResultOk, msg, callback)); - } else { - // Regular path, append individual message to incoming messages queue - incomingMessages_.push(msg); - lock.unlock(); - } + executeNotifyCallback(msg); } if (skippedMessages > 0) { @@ -685,7 +691,7 @@ void ConsumerImpl::internalListener() { try { consumerStatsBasePtr_->receivedMessage(msg, ResultOk); lastDequedMessageId_ = msg.getMessageId(); - messageListener_(Consumer(shared_from_this()), msg); + messageListener_(Consumer(get_shared_this_ptr()), msg); } catch (const std::exception& e) { LOG_ERROR(getName() << "Exception thrown from listener" << e.what()); } @@ -708,9 +714,7 @@ Result ConsumerImpl::fetchSingleMessageFromBroker(Message& msg) { getName() << "The incoming message queue should never be greater than 0 when Queue size is 0"); incomingMessages_.clear(); } - Lock localLock(mutex_); waitingForZeroQueueSizeMessage = true; - localLock.unlock(); sendFlowPermitsToBroker(currentCnx, 1); @@ -732,7 +736,6 @@ Result ConsumerImpl::fetchSingleMessageFromBroker(Message& msg) { } } } - return ResultOk; } Result ConsumerImpl::receive(Message& msg) { @@ -824,6 +827,8 @@ void ConsumerImpl::messageProcessed(Message& msg, bool track) { lastDequedMessageId_ = msg.getMessageId(); lock.unlock(); + incomingMessagesSize_.fetch_sub(msg.getLength()); + ClientConnectionPtr currentCnx = getCnx().lock(); if (currentCnx && msg.impl_->cnx_ != currentCnx.get()) { LOG_DEBUG(getName() << "Not adding permit since connection is different."); @@ -915,7 +920,7 @@ void ConsumerImpl::statsCallback(Result res, ResultCallback callback, proto::Com } void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callback) { - ResultCallback cb = std::bind(&ConsumerImpl::statsCallback, shared_from_this(), std::placeholders::_1, + ResultCallback cb = std::bind(&ConsumerImpl::statsCallback, get_shared_this_ptr(), std::placeholders::_1, callback, proto::CommandAck_AckType_Individual); if (msgId.batchIndex() != -1 && !batchAcknowledgementTracker_.isBatchReady(msgId, proto::CommandAck_AckType_Individual)) { @@ -926,7 +931,7 @@ void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callb } void ConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) { - ResultCallback cb = std::bind(&ConsumerImpl::statsCallback, shared_from_this(), std::placeholders::_1, + ResultCallback cb = std::bind(&ConsumerImpl::statsCallback, get_shared_this_ptr(), std::placeholders::_1, callback, proto::CommandAck_AckType_Cumulative); if (!isCumulativeAcknowledgementAllowed(config_.getConsumerType())) { cb(ResultCumulativeAcknowledgementNotAllowedError); @@ -974,12 +979,12 @@ void ConsumerImpl::disconnectConsumer() { Lock lock(mutex_); connection_.reset(); lock.unlock(); - scheduleReconnection(shared_from_this()); + scheduleReconnection(get_shared_this_ptr()); } void ConsumerImpl::closeAsync(ResultCallback callback) { // Keep a reference to ensure object is kept alive - ConsumerImplPtr ptr = shared_from_this(); + ConsumerImplPtr ptr = get_shared_this_ptr(); if (state_ != Ready) { if (callback) { @@ -1022,12 +1027,16 @@ void ConsumerImpl::closeAsync(ResultCallback callback) { cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId); if (callback) { // Pass the shared pointer "ptr" to the handler to prevent the object from being destroyed - future.addListener( - std::bind(&ConsumerImpl::handleClose, shared_from_this(), std::placeholders::_1, callback, ptr)); + future.addListener(std::bind(&ConsumerImpl::handleClose, get_shared_this_ptr(), std::placeholders::_1, + callback, ptr)); } // fail pendingReceive callback failPendingReceiveCallback(); + failPendingBatchReceiveCallback(); + + // cancel timer + batchReceiveTimer_->cancel(); } void ConsumerImpl::handleClose(Result result, ResultCallback callback, ConsumerImplPtr consumer) { @@ -1083,7 +1092,7 @@ Result ConsumerImpl::resumeMessageListener() { for (size_t i = 0; i < count; i++) { // Trigger message listener callback in a separate thread - listenerExecutor_->postWork(std::bind(&ConsumerImpl::internalListener, shared_from_this())); + listenerExecutor_->postWork(std::bind(&ConsumerImpl::internalListener, get_shared_this_ptr())); } // Check current permits and determine whether to send FLOW command this->increaseAvailablePermits(getCnx().lock(), 0); @@ -1148,7 +1157,7 @@ void ConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callb << ", requestId - " << requestId); cnx->newConsumerStats(consumerId_, requestId) - .addListener(std::bind(&ConsumerImpl::brokerConsumerStatsListener, shared_from_this(), + .addListener(std::bind(&ConsumerImpl::brokerConsumerStatsListener, get_shared_this_ptr(), std::placeholders::_1, std::placeholders::_2, callback)); return; } else { @@ -1209,7 +1218,7 @@ void ConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) { if (callback) { future.addListener( - std::bind(&ConsumerImpl::handleSeek, shared_from_this(), std::placeholders::_1, callback)); + std::bind(&ConsumerImpl::handleSeek, get_shared_this_ptr(), std::placeholders::_1, callback)); } return; } @@ -1239,7 +1248,7 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) { if (callback) { future.addListener( - std::bind(&ConsumerImpl::handleSeek, shared_from_this(), std::placeholders::_1, callback)); + std::bind(&ConsumerImpl::handleSeek, get_shared_this_ptr(), std::placeholders::_1, callback)); } return; } @@ -1317,7 +1326,7 @@ void ConsumerImpl::internalGetLastMessageIdAsync(const BackoffPtr& backoff, Time LOG_DEBUG(getName() << " Sending getLastMessageId Command for Consumer - " << getConsumerId() << ", requestId - " << requestId); - auto self = shared_from_this(); + auto self = get_shared_this_ptr(); cnx->newGetLastMessageId(consumerId_, requestId) .addListener([this, self, callback](Result result, const GetLastMessageIdResponse& response) { if (result == ResultOk) { @@ -1380,4 +1389,18 @@ bool ConsumerImpl::isConnected() const { return !getCnx().expired() && state_ == uint64_t ConsumerImpl::getNumberOfConnectedConsumer() { return isConnected() ? 1 : 0; } +bool ConsumerImpl::hasEnoughMessagesForBatchReceive() const { + if (batchReceivePolicy_.getMaxNumMessages() <= 0 && batchReceivePolicy_.getMaxNumBytes() <= 0) { + return false; + } + return (batchReceivePolicy_.getMaxNumMessages() > 0 && + incomingMessages_.size() >= batchReceivePolicy_.getMaxNumMessages()) || + (batchReceivePolicy_.getMaxNumBytes() > 0 && + incomingMessagesSize_ >= batchReceivePolicy_.getMaxNumBytes()); +} + +std::shared_ptr ConsumerImpl::get_shared_this_ptr() { + return std::dynamic_pointer_cast(shared_from_this()); +} + } /* namespace pulsar */ diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 70fda0170cc1a..cb336fb9208c8 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -64,9 +64,7 @@ enum ConsumerTopicType Partitioned }; -class ConsumerImpl : public ConsumerImplBase, - public HandlerBase, - public std::enable_shared_from_this { +class ConsumerImpl : public ConsumerImplBase { public: ConsumerImpl(const ClientImplPtr client, const std::string& topic, const std::string& subscriptionName, const ConsumerConfiguration&, @@ -147,7 +145,10 @@ class ConsumerImpl : public ConsumerImplBase, // overrided methods from HandlerBase void connectionOpened(const ClientConnectionPtr& cnx) override; void connectionFailed(Result result) override; - HandlerBaseWeakPtr get_weak_from_this() override { return shared_from_this(); } + + // impl methods from ConsumerImpl base + bool hasEnoughMessagesForBatchReceive() const override; + void notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) override; void handleCreateConsumer(const ClientConnectionPtr& cnx, Result result); @@ -159,7 +160,8 @@ class ConsumerImpl : public ConsumerImplBase, ConsumerStatsBasePtr consumerStatsBasePtr_; private: - bool waitingForZeroQueueSizeMessage; + volatile bool waitingForZeroQueueSizeMessage; + std::shared_ptr get_shared_this_ptr(); bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData, const proto::MessageMetadata& metadata, SharedBuffer& payload, bool checkMaxMessageSize); @@ -178,6 +180,7 @@ class ConsumerImpl : public ConsumerImplBase, Result receiveHelper(Message& msg); Result receiveHelper(Message& msg, int timeout); void statsCallback(Result, ResultCallback, proto::CommandAck_AckType); + void executeNotifyCallback(Message& msg); void notifyPendingReceivedCallback(Result result, Message& message, const ReceiveCallback& callback); void failPendingReceiveCallback(); void setNegativeAcknowledgeEnabledForTesting(bool enabled) override; @@ -194,13 +197,13 @@ class ConsumerImpl : public ConsumerImplBase, std::string originalSubscriptionName_; MessageListener messageListener_; ConsumerEventListenerPtr eventListener_; - ExecutorServicePtr listenerExecutor_; bool hasParent_; ConsumerTopicType consumerTopicType_; const Commands::SubscriptionMode subscriptionMode_; UnboundedBlockingQueue incomingMessages_; + std::atomic_int incomingMessagesSize_ = {0}; std::queue pendingReceives_; std::atomic_int availablePermits_; const int receiverQueueRefillThreshold_; diff --git a/pulsar-client-cpp/lib/ConsumerImplBase.cc b/pulsar-client-cpp/lib/ConsumerImplBase.cc new file mode 100644 index 0000000000000..73fc75f022878 --- /dev/null +++ b/pulsar-client-cpp/lib/ConsumerImplBase.cc @@ -0,0 +1,144 @@ +/** + * 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 "ConsumerImpl.h" +#include "MessageImpl.h" +#include "MessagesImpl.h" +#include "LogUtils.h" +#include "TimeUtils.h" +#include "pulsar/Result.h" +#include "MessageIdUtil.h" +#include "AckGroupingTracker.h" +#include "ConsumerImplBase.h" + +#include + +DECLARE_LOG_OBJECT() + +namespace pulsar { + +ConsumerImplBase::ConsumerImplBase(ClientImplPtr client, const std::string& topic, Backoff backoff, + const ConsumerConfiguration& conf, ExecutorServicePtr listenerExecutor) + : HandlerBase(client, topic, backoff), + listenerExecutor_(listenerExecutor), + batchReceivePolicy_(conf.getBatchReceivePolicy()) { + auto userBatchReceivePolicy = conf.getBatchReceivePolicy(); + if (userBatchReceivePolicy.getMaxNumMessages() > conf.getReceiverQueueSize()) { + batchReceivePolicy_ = + BatchReceivePolicy(conf.getReceiverQueueSize(), userBatchReceivePolicy.getMaxNumBytes(), + userBatchReceivePolicy.getTimeoutMs()); + LOG_WARN("BatchReceivePolicy maxNumMessages: {" << userBatchReceivePolicy.getMaxNumMessages() + << "} is greater than maxReceiverQueueSize: {" + << conf.getReceiverQueueSize() + << "}, reset to " + "maxReceiverQueueSize. "); + } + batchReceiveTimer_ = listenerExecutor_->createDeadlineTimer(); +} + +void ConsumerImplBase::triggerBatchReceiveTimerTask(long timeoutMs) { + if (timeoutMs > 0) { + batchReceiveTimer_->expires_from_now(boost::posix_time::milliseconds(timeoutMs)); + auto self = shared_from_this(); + batchReceiveTimer_->async_wait([self](const boost::system::error_code& ec) { + // If two requests call runPartitionUpdateTask at the same time, the timer will fail, and it + // cannot continue at this time, and the request needs to be ignored. + if (!ec) { + self->doBatchReceiveTimeTask(); + } + }); + } +} + +void ConsumerImplBase::doBatchReceiveTimeTask() { + if (state_ != Ready) { + return; + } + + bool hasPendingReceives = false; + long timeToWaitMs = batchReceivePolicy_.getTimeoutMs(); + + Lock lock(batchPendingReceiveMutex_); + while (!batchPendingReceives_.empty()) { + OpBatchReceive& batchReceive = batchPendingReceives_.front(); + long diff = + batchReceivePolicy_.getTimeoutMs() - (TimeUtils::currentTimeMillis() - batchReceive.createAt_); + if (diff <= 0) { + notifyBatchPendingReceivedCallback(batchReceive.batchReceiveCallback_); + batchPendingReceives_.pop(); + } else { + hasPendingReceives = true; + timeToWaitMs = diff; + break; + } + } + lock.unlock(); + + if (hasPendingReceives) { + triggerBatchReceiveTimerTask(timeToWaitMs); + } +} + +void ConsumerImplBase::failPendingBatchReceiveCallback() { + Messages msgs; + Lock lock(batchPendingReceiveMutex_); + while (!batchPendingReceives_.empty()) { + OpBatchReceive opBatchReceive = batchPendingReceives_.front(); + batchPendingReceives_.pop(); + auto self = shared_from_this(); + listenerExecutor_->postWork([opBatchReceive, self, msgs]() { + opBatchReceive.batchReceiveCallback_(ResultAlreadyClosed, msgs); + }); + } + lock.unlock(); +} + +void ConsumerImplBase::notifyBatchPendingReceivedCallback() { + Lock lock(batchPendingReceiveMutex_); + if (!batchPendingReceives_.empty()) { + OpBatchReceive& batchReceive = batchPendingReceives_.front(); + batchPendingReceives_.pop(); + notifyBatchPendingReceivedCallback(batchReceive.batchReceiveCallback_); + } +} + +void ConsumerImplBase::batchReceiveAsync(BatchReceiveCallback callback) { + // fail the callback if consumer is closing or closed + if (state_ != Ready) { + callback(ResultAlreadyClosed, Messages()); + return; + } + + if (hasEnoughMessagesForBatchReceive()) { + Lock lock(batchPendingReceiveMutex_); + notifyBatchPendingReceivedCallback(callback); + lock.unlock(); + } else { + // expectmoreIncomingMessages(); + OpBatchReceive opBatchReceive(callback); + Lock lock(batchPendingReceiveMutex_); + batchPendingReceives_.emplace(opBatchReceive); + lock.unlock(); + triggerBatchReceiveTimerTask(batchReceivePolicy_.getTimeoutMs()); + } +} + +OpBatchReceive::OpBatchReceive(const BatchReceiveCallback& batchReceiveCallback) + : batchReceiveCallback_(batchReceiveCallback), createAt_(TimeUtils::currentTimeMillis()) {} + +} /* namespace pulsar */ diff --git a/pulsar-client-cpp/lib/ConsumerImplBase.h b/pulsar-client-cpp/lib/ConsumerImplBase.h index 693d4da9a3779..dc328606a0a28 100644 --- a/pulsar-client-cpp/lib/ConsumerImplBase.h +++ b/pulsar-client-cpp/lib/ConsumerImplBase.h @@ -20,23 +20,38 @@ #define PULSAR_CONSUMER_IMPL_BASE_HEADER #include #include - +#include "HandlerBase.h" +#include #include namespace pulsar { class ConsumerImplBase; +class HandlerBase; typedef std::weak_ptr ConsumerImplBaseWeakPtr; -class ConsumerImplBase { +class OpBatchReceive { public: - virtual ~ConsumerImplBase() {} + OpBatchReceive(); + explicit OpBatchReceive(const BatchReceiveCallback& batchReceiveCallback); + const BatchReceiveCallback batchReceiveCallback_; + const long createAt_; +}; + +class ConsumerImplBase : public HandlerBase, public std::enable_shared_from_this { + public: + virtual ~ConsumerImplBase(){}; + ConsumerImplBase(ClientImplPtr client, const std::string& topic, Backoff backoff, + const ConsumerConfiguration& conf, ExecutorServicePtr listenerExecutor); + + // interface by consumer virtual Future getConsumerCreatedFuture() = 0; - virtual const std::string& getSubscriptionName() const = 0; virtual const std::string& getTopic() const = 0; + virtual const std::string& getSubscriptionName() const = 0; virtual Result receive(Message& msg) = 0; virtual Result receive(Message& msg, int timeout) = 0; virtual void receiveAsync(ReceiveCallback& callback) = 0; + void batchReceiveAsync(BatchReceiveCallback callback); virtual void unsubscribeAsync(ResultCallback callback) = 0; virtual void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) = 0; virtual void acknowledgeCumulativeAsync(const MessageId& msgId, ResultCallback callback) = 0; @@ -49,7 +64,6 @@ class ConsumerImplBase { virtual Result resumeMessageListener() = 0; virtual void redeliverUnacknowledgedMessages() = 0; virtual void redeliverUnacknowledgedMessages(const std::set& messageIds) = 0; - virtual const std::string& getName() const = 0; virtual int getNumOfPrefetchedMessages() const = 0; virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) = 0; virtual void seekAsync(const MessageId& msgId, ResultCallback callback) = 0; @@ -57,6 +71,26 @@ class ConsumerImplBase { virtual void negativeAcknowledge(const MessageId& msgId) = 0; virtual bool isConnected() const = 0; virtual uint64_t getNumberOfConnectedConsumer() = 0; + // overrided methods from HandlerBase + virtual const std::string& getName() const override = 0; + + protected: + // overrided methods from HandlerBase + void connectionOpened(const ClientConnectionPtr& cnx) override {} + void connectionFailed(Result result) override {} + HandlerBaseWeakPtr get_weak_from_this() override { return shared_from_this(); } + + // consumer impl generic method. + ExecutorServicePtr listenerExecutor_; + std::queue batchPendingReceives_; + BatchReceivePolicy batchReceivePolicy_; + DeadlineTimerPtr batchReceiveTimer_; + void triggerBatchReceiveTimerTask(long timeoutMs); + void doBatchReceiveTimeTask(); + void failPendingBatchReceiveCallback(); + void notifyBatchPendingReceivedCallback(); + virtual void notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) = 0; + virtual bool hasEnoughMessagesForBatchReceive() const = 0; private: virtual void setNegativeAcknowledgeEnabledForTesting(bool enabled) = 0; diff --git a/pulsar-client-cpp/lib/HandlerBase.h b/pulsar-client-cpp/lib/HandlerBase.h index 1184746da21ba..6fc3603dbdcf4 100644 --- a/pulsar-client-cpp/lib/HandlerBase.h +++ b/pulsar-client-cpp/lib/HandlerBase.h @@ -90,6 +90,7 @@ class HandlerBase { ExecutorServicePtr executor_; mutable std::mutex mutex_; std::mutex pendingReceiveMutex_; + std::mutex batchPendingReceiveMutex_; ptime creationTimestamp_; const TimeDuration operationTimeut_; diff --git a/pulsar-client-cpp/lib/Messages.cc b/pulsar-client-cpp/lib/Messages.cc new file mode 100644 index 0000000000000..fee2ba5e2d479 --- /dev/null +++ b/pulsar-client-cpp/lib/Messages.cc @@ -0,0 +1,34 @@ +/** + * 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 +#include +#include "MessagesImpl.h" + +using namespace pulsar; + +namespace pulsar { + +Messages::Messages() : impl_() {} + +std::vector Messages::getMessageList() const { return impl_->getMessageList(); } + +Messages::Messages(Messages::MessagesImplPtr msgsPtr) : impl_(msgsPtr) {} + +} // namespace pulsar diff --git a/pulsar-client-cpp/lib/MessagesImpl.cc b/pulsar-client-cpp/lib/MessagesImpl.cc new file mode 100644 index 0000000000000..15d321a484c86 --- /dev/null +++ b/pulsar-client-cpp/lib/MessagesImpl.cc @@ -0,0 +1,62 @@ +/** + * 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 "MessagesImpl.h" + +MessagesImpl::MessagesImpl(int maxNumberOfMessages, long maxSizeOfMessages) + : maxNumberOfMessages_(maxNumberOfMessages), + maxSizeOfMessages_(maxSizeOfMessages), + currentNumberOfMessages_(0), + currentSizeOfMessages_(0) { + messageList_ = std::vector(); +} + +const std::vector& MessagesImpl::getMessageList() const { return messageList_; } + +bool MessagesImpl::canAdd(const Message& message) const { + if (currentNumberOfMessages_ == 0) { + return true; + } + + if (maxNumberOfMessages_ > 0 && currentNumberOfMessages_ + 1 > maxNumberOfMessages_) { + return false; + } + + if (maxSizeOfMessages_ > 0 && currentSizeOfMessages_ + message.getLength() > maxSizeOfMessages_) { + return false; + } + + return true; +} + +void MessagesImpl::add(const Message& message) { + if (!canAdd(message)) { + throw std::invalid_argument("No more space to add messages."); + } + currentNumberOfMessages_++; + currentSizeOfMessages_ += message.getLength(); + messageList_.emplace_back(message); +} + +int MessagesImpl::size() const { return messageList_.size(); } + +void MessagesImpl::clear() { + currentNumberOfMessages_ = 0; + currentSizeOfMessages_ = 0; + messageList_.clear(); +} diff --git a/pulsar-client-cpp/lib/MessagesImpl.h b/pulsar-client-cpp/lib/MessagesImpl.h new file mode 100644 index 0000000000000..1c4d556c6d76d --- /dev/null +++ b/pulsar-client-cpp/lib/MessagesImpl.h @@ -0,0 +1,48 @@ +/** + * 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. + */ +#ifndef PULSAR_CPP_MESSAGESIMPL_H +#define PULSAR_CPP_MESSAGESIMPL_H + +#include +#include +#include + +using namespace pulsar; + +namespace pulsar { + +class MessagesImpl { + public: + MessagesImpl(const int maxNumberOfMessages, const long maxSizeOfMessages); + const std::vector& getMessageList() const; + bool canAdd(const Message& message) const; + void add(const Message& message); + int size() const; + void clear(); + + private: + std::vector messageList_; + const int maxNumberOfMessages_; + const long maxSizeOfMessages_; + int currentNumberOfMessages_; + long currentSizeOfMessages_; +}; + +} // namespace pulsar +#endif // PULSAR_CPP_MESSAGESIMPL_H diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc index 7515076234556..5c7bb929aa85d 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc @@ -18,6 +18,7 @@ */ #include "MultiTopicsConsumerImpl.h" #include "MultiResultCallback.h" +#include "MessagesImpl.h" DECLARE_LOG_OBJECT() @@ -27,12 +28,13 @@ MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, const std const std::string& subscriptionName, TopicNamePtr topicName, const ConsumerConfiguration& conf, LookupServicePtr lookupServicePtr) - : client_(client), + : ConsumerImplBase(client, topicName ? topicName->toString() : "EmptyTopics", + Backoff(milliseconds(100), seconds(60), milliseconds(0)), conf, + client->getListenerExecutorProvider()->get()), + client_(client), subscriptionName_(subscriptionName), - topic_(topicName ? topicName->toString() : "EmptyTopics"), conf_(conf), - messages_(conf.getReceiverQueueSize()), - listenerExecutor_(client->getListenerExecutorProvider()->get()), + incomingMessages_(conf.getReceiverQueueSize()), messageListener_(conf.getMessageListener()), lookupServicePtr_(lookupServicePtr), numberTopicPartitions_(std::make_shared>(0)), @@ -59,14 +61,16 @@ MultiTopicsConsumerImpl::MultiTopicsConsumerImpl(ClientImplPtr client, const std partitionsUpdateInterval_ = boost::posix_time::seconds(partitionsUpdateInterval); lookupServicePtr_ = client_->getLookup(); } + + state_ = Pending; } void MultiTopicsConsumerImpl::start() { if (topics_.empty()) { - MultiTopicsConsumerState state = Pending; + State state = Pending; if (state_.compare_exchange_strong(state, Ready)) { LOG_DEBUG("No topics passed in when create MultiTopicsConsumer."); - multiTopicsConsumerCreatedPromise_.setValue(shared_from_this()); + multiTopicsConsumerCreatedPromise_.setValue(get_shared_this_ptr()); return; } else { LOG_ERROR("Consumer " << consumerStr_ << " in wrong state: " << state_); @@ -81,7 +85,7 @@ void MultiTopicsConsumerImpl::start() { // subscribe for each passed in topic for (std::vector::const_iterator itr = topics_.begin(); itr != topics_.end(); itr++) { subscribeOneTopicAsync(*itr).addListener(std::bind(&MultiTopicsConsumerImpl::handleOneTopicSubscribed, - shared_from_this(), std::placeholders::_1, + get_shared_this_ptr(), std::placeholders::_1, std::placeholders::_2, *itr, topicsNeedCreate)); } } @@ -100,10 +104,10 @@ void MultiTopicsConsumerImpl::handleOneTopicSubscribed(Result result, Consumer c } if (--(*topicsNeedCreate) == 0) { - MultiTopicsConsumerState state = Pending; + State state = Pending; if (state_.compare_exchange_strong(state, Ready)) { LOG_INFO("Successfully Subscribed to Topics"); - multiTopicsConsumerCreatedPromise_.setValue(shared_from_this()); + multiTopicsConsumerCreatedPromise_.setValue(get_shared_this_ptr()); } else { LOG_ERROR("Unable to create Consumer - " << consumerStr_ << " Error - " << result); // unsubscribed all of the successfully subscribed partitioned consumers @@ -162,7 +166,7 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicN ConsumerConfiguration config = conf_.clone(); ExecutorServicePtr internalListenerExecutor = client_->getPartitionListenerExecutorProvider()->get(); - config.setMessageListener(std::bind(&MultiTopicsConsumerImpl::messageReceived, shared_from_this(), + config.setMessageListener(std::bind(&MultiTopicsConsumerImpl::messageReceived, get_shared_this_ptr(), std::placeholders::_1, std::placeholders::_2)); int partitions = numPartitions == 0 ? 1 : numPartitions; @@ -185,8 +189,8 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicN consumer = std::make_shared(client_, topicName->toString(), subscriptionName_, config, internalListenerExecutor, true, NonPartitioned); consumer->getConsumerCreatedFuture().addListener(std::bind( - &MultiTopicsConsumerImpl::handleSingleConsumerCreated, shared_from_this(), std::placeholders::_1, - std::placeholders::_2, partitionsNeedCreate, topicSubResultPromise)); + &MultiTopicsConsumerImpl::handleSingleConsumerCreated, get_shared_this_ptr(), + std::placeholders::_1, std::placeholders::_2, partitionsNeedCreate, topicSubResultPromise)); consumers_.emplace(topicName->toString(), consumer); LOG_DEBUG("Creating Consumer for - " << topicName << " - " << consumerStr_); consumer->start(); @@ -197,7 +201,7 @@ void MultiTopicsConsumerImpl::subscribeTopicPartitions(int numPartitions, TopicN consumer = std::make_shared(client_, topicPartitionName, subscriptionName_, config, internalListenerExecutor, true, Partitioned); consumer->getConsumerCreatedFuture().addListener(std::bind( - &MultiTopicsConsumerImpl::handleSingleConsumerCreated, shared_from_this(), + &MultiTopicsConsumerImpl::handleSingleConsumerCreated, get_shared_this_ptr(), std::placeholders::_1, std::placeholders::_2, partitionsNeedCreate, topicSubResultPromise)); consumer->setPartitionIndex(i); consumers_.emplace(topicPartitionName, consumer); @@ -234,7 +238,7 @@ void MultiTopicsConsumerImpl::handleSingleConsumerCreated( if (partitionsUpdateTimer_) { runPartitionUpdateTask(); } - topicSubResultPromise->setValue(Consumer(shared_from_this())); + topicSubResultPromise->setValue(Consumer(get_shared_this_ptr())); } } @@ -250,7 +254,7 @@ void MultiTopicsConsumerImpl::unsubscribeAsync(ResultCallback callback) { state_ = Closing; std::shared_ptr> consumerUnsubed = std::make_shared>(0); - auto self = shared_from_this(); + auto self = get_shared_this_ptr(); int numConsumers = 0; consumers_.forEachValue( [&numConsumers, &consumerUnsubed, &self, callback](const ConsumerImplPtr& consumer) { @@ -327,7 +331,7 @@ void MultiTopicsConsumerImpl::unsubscribeOneTopicAsync(const std::string& topic, } optConsumer.value()->unsubscribeAsync( - std::bind(&MultiTopicsConsumerImpl::handleOneTopicUnsubscribedAsync, shared_from_this(), + std::bind(&MultiTopicsConsumerImpl::handleOneTopicUnsubscribedAsync, get_shared_this_ptr(), std::placeholders::_1, consumerUnsubed, numberPartitions, topicName, topicPartitionName, callback)); } @@ -383,7 +387,7 @@ void MultiTopicsConsumerImpl::closeAsync(ResultCallback callback) { state_ = Closing; - std::weak_ptr weakSelf{shared_from_this()}; + std::weak_ptr weakSelf{get_shared_this_ptr()}; int numConsumers = 0; consumers_.clear( [this, weakSelf, &numConsumers, callback](const std::string& name, const ConsumerImplPtr& consumer) { @@ -412,7 +416,7 @@ void MultiTopicsConsumerImpl::closeAsync(ResultCallback callback) { } // closed all consumers if (numConsumersLeft == 0) { - messages_.clear(); + incomingMessages_.clear(); topicsPartitions_.clear(); unAckedMessageTrackerPtr_->clear(); @@ -438,6 +442,10 @@ void MultiTopicsConsumerImpl::closeAsync(ResultCallback callback) { // fail pending receive failPendingReceiveCallback(); + failPendingBatchReceiveCallback(); + + // cancel timer + batchReceiveTimer_->cancel(); } void MultiTopicsConsumerImpl::messageReceived(Consumer consumer, const Message& msg) { @@ -452,25 +460,37 @@ void MultiTopicsConsumerImpl::messageReceived(Consumer consumer, const Message& pendingReceives_.pop(); lock.unlock(); listenerExecutor_->postWork(std::bind(&MultiTopicsConsumerImpl::notifyPendingReceivedCallback, - shared_from_this(), ResultOk, msg, callback)); - } else { - if (messages_.full()) { - lock.unlock(); - } + get_shared_this_ptr(), ResultOk, msg, callback)); + return; + } - if (messages_.push(msg) && messageListener_) { - listenerExecutor_->postWork( - std::bind(&MultiTopicsConsumerImpl::internalListener, shared_from_this(), consumer)); - } + if (incomingMessages_.full()) { + lock.unlock(); + } + + // add message to block queue. + // when messages queue is full, will block listener thread on ConsumerImpl, + // then will not send permits to broker, will broker stop push message. + incomingMessages_.push(msg); + incomingMessagesSize_.fetch_add(msg.getLength()); + + // try trigger pending batch messages + if (hasEnoughMessagesForBatchReceive()) { + ConsumerImplBase::notifyBatchPendingReceivedCallback(); + } + + if (messageListener_) { + listenerExecutor_->postWork( + std::bind(&MultiTopicsConsumerImpl::internalListener, get_shared_this_ptr(), consumer)); } } void MultiTopicsConsumerImpl::internalListener(Consumer consumer) { Message m; - messages_.pop(m); - unAckedMessageTrackerPtr_->add(m.getMessageId()); + incomingMessages_.pop(m); try { - messageListener_(Consumer(shared_from_this()), m); + messageListener_(Consumer(get_shared_this_ptr()), m); + messageProcessed(m); } catch (const std::exception& e) { LOG_ERROR("Exception thrown from listener of Partitioned Consumer" << e.what()); } @@ -485,9 +505,9 @@ Result MultiTopicsConsumerImpl::receive(Message& msg) { LOG_ERROR("Can not receive when a listener has been set"); return ResultInvalidConfiguration; } - messages_.pop(msg); + incomingMessages_.pop(msg); + messageProcessed(msg); - unAckedMessageTrackerPtr_->add(msg.getMessageId()); return ResultOk; } @@ -501,8 +521,8 @@ Result MultiTopicsConsumerImpl::receive(Message& msg, int timeout) { return ResultInvalidConfiguration; } - if (messages_.pop(msg, std::chrono::milliseconds(timeout))) { - unAckedMessageTrackerPtr_->add(msg.getMessageId()); + if (incomingMessages_.pop(msg, std::chrono::milliseconds(timeout))) { + messageProcessed(msg); return ResultOk; } else { if (state_ != Ready) { @@ -522,9 +542,9 @@ void MultiTopicsConsumerImpl::receiveAsync(ReceiveCallback& callback) { } Lock lock(pendingReceiveMutex_); - if (messages_.pop(msg, std::chrono::milliseconds(0))) { + if (incomingMessages_.pop(msg, std::chrono::milliseconds(0))) { lock.unlock(); - unAckedMessageTrackerPtr_->add(msg.getMessageId()); + messageProcessed(msg); callback(ResultOk, msg); } else { pendingReceives_.push(callback); @@ -534,14 +554,14 @@ void MultiTopicsConsumerImpl::receiveAsync(ReceiveCallback& callback) { void MultiTopicsConsumerImpl::failPendingReceiveCallback() { Message msg; - messages_.close(); + incomingMessages_.close(); Lock lock(pendingReceiveMutex_); while (!pendingReceives_.empty()) { ReceiveCallback callback = pendingReceives_.front(); pendingReceives_.pop(); listenerExecutor_->postWork(std::bind(&MultiTopicsConsumerImpl::notifyPendingReceivedCallback, - shared_from_this(), ResultAlreadyClosed, msg, callback)); + get_shared_this_ptr(), ResultAlreadyClosed, msg, callback)); } lock.unlock(); } @@ -647,7 +667,7 @@ void MultiTopicsConsumerImpl::redeliverUnacknowledgedMessages(const std::set(numberTopicPartitions_->load()); lock.unlock(); - auto self = shared_from_this(); + auto self = get_shared_this_ptr(); size_t i = 0; consumers_.forEachValue([&self, &latchPtr, &statsPtr, &i, callback](const ConsumerImplPtr& consumer) { size_t index = i++; @@ -748,7 +768,7 @@ uint64_t MultiTopicsConsumerImpl::getNumberOfConnectedConsumer() { } void MultiTopicsConsumerImpl::runPartitionUpdateTask() { partitionsUpdateTimer_->expires_from_now(partitionsUpdateInterval_); - std::weak_ptr weakSelf{shared_from_this()}; + std::weak_ptr weakSelf{get_shared_this_ptr()}; partitionsUpdateTimer_->async_wait([weakSelf](const boost::system::error_code& ec) { // If two requests call runPartitionUpdateTask at the same time, the timer will fail, and it // cannot continue at this time, and the request needs to be ignored. @@ -767,7 +787,7 @@ void MultiTopicsConsumerImpl::topicPartitionUpdate() { auto topicName = TopicName::get(item.first); auto currentNumPartitions = item.second; lookupServicePtr_->getPartitionMetadataAsync(topicName).addListener( - std::bind(&MultiTopicsConsumerImpl::handleGetPartitions, shared_from_this(), topicName, + std::bind(&MultiTopicsConsumerImpl::handleGetPartitions, get_shared_this_ptr(), topicName, std::placeholders::_1, std::placeholders::_2, currentNumPartitions)); } } @@ -808,7 +828,7 @@ void MultiTopicsConsumerImpl::subscribeSingleNewConsumer( std::shared_ptr> partitionsNeedCreate) { ConsumerConfiguration config = conf_.clone(); ExecutorServicePtr internalListenerExecutor = client_->getPartitionListenerExecutorProvider()->get(); - config.setMessageListener(std::bind(&MultiTopicsConsumerImpl::messageReceived, shared_from_this(), + config.setMessageListener(std::bind(&MultiTopicsConsumerImpl::messageReceived, get_shared_this_ptr(), std::placeholders::_1, std::placeholders::_2)); // Apply total limit of receiver queue size across partitions @@ -821,7 +841,7 @@ void MultiTopicsConsumerImpl::subscribeSingleNewConsumer( auto consumer = std::make_shared(client_, topicPartitionName, subscriptionName_, config, internalListenerExecutor, true, Partitioned); consumer->getConsumerCreatedFuture().addListener( - std::bind(&MultiTopicsConsumerImpl::handleSingleConsumerCreated, shared_from_this(), + std::bind(&MultiTopicsConsumerImpl::handleSingleConsumerCreated, get_shared_this_ptr(), std::placeholders::_1, std::placeholders::_2, partitionsNeedCreate, topicSubResultPromise)); consumer->setPartitionIndex(partitionIndex); consumer->start(); @@ -829,3 +849,41 @@ void MultiTopicsConsumerImpl::subscribeSingleNewConsumer( LOG_INFO("Add Creating Consumer for - " << topicPartitionName << " - " << consumerStr_ << " consumerSize: " << consumers_.size()); } + +bool MultiTopicsConsumerImpl::hasEnoughMessagesForBatchReceive() const { + if (batchReceivePolicy_.getMaxNumMessages() <= 0 && batchReceivePolicy_.getMaxNumBytes() <= 0) { + return false; + } + return (batchReceivePolicy_.getMaxNumMessages() > 0 && + incomingMessages_.size() >= batchReceivePolicy_.getMaxNumMessages()) || + (batchReceivePolicy_.getMaxNumBytes() > 0 && + incomingMessagesSize_ >= batchReceivePolicy_.getMaxNumBytes()); +} + +void MultiTopicsConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) { + auto messages = std::make_shared(batchReceivePolicy_.getMaxNumMessages(), + batchReceivePolicy_.getMaxNumBytes()); + + Message peekMsg; + while (incomingMessages_.peek(peekMsg) && messages->canAdd(peekMsg)) { + // decreaseIncomingMessageSize + Message msg; + incomingMessages_.pop(msg); + messageProcessed(msg); + messages->add(msg); + } + auto self = get_shared_this_ptr(); + listenerExecutor_->postWork([callback, messages, self]() { + Messages msgs(messages); + callback(ResultOk, msgs); + }); +} + +void MultiTopicsConsumerImpl::messageProcessed(Message& msg) { + incomingMessagesSize_.fetch_sub(msg.getLength()); + unAckedMessageTrackerPtr_->add(msg.getMessageId()); +} + +std::shared_ptr MultiTopicsConsumerImpl::get_shared_this_ptr() { + return std::dynamic_pointer_cast(shared_from_this()); +} diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h index 8769d59b9908e..044f4173b6885 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h @@ -38,17 +38,8 @@ namespace pulsar { typedef std::shared_ptr> ConsumerSubResultPromisePtr; class MultiTopicsConsumerImpl; -class MultiTopicsConsumerImpl : public ConsumerImplBase, - public std::enable_shared_from_this { +class MultiTopicsConsumerImpl : public ConsumerImplBase { public: - enum MultiTopicsConsumerState - { - Pending, - Ready, - Closing, - Closed, - Failed - }; MultiTopicsConsumerImpl(ClientImplPtr client, const std::vector& topics, const std::string& subscriptionName, TopicNamePtr topicName, const ConsumerConfiguration& conf, LookupServicePtr lookupServicePtr_); @@ -99,16 +90,14 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase, const ClientImplPtr client_; const std::string subscriptionName_; std::string consumerStr_; - std::string topic_; const ConsumerConfiguration conf_; typedef SynchronizedHashMap ConsumerMap; ConsumerMap consumers_; std::map topicsPartitions_; mutable std::mutex mutex_; std::mutex pendingReceiveMutex_; - std::atomic state_{Pending}; - BlockingQueue messages_; - const ExecutorServicePtr listenerExecutor_; + BlockingQueue incomingMessages_; + std::atomic_int incomingMessagesSize_ = {0}; MessageListener messageListener_; DeadlineTimerPtr partitionsUpdateTimer_; boost::posix_time::time_duration partitionsUpdateInterval_; @@ -125,6 +114,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase, unsigned int partitionIndex); void notifyResult(CloseCallback closeCallback); void messageReceived(Consumer consumer, const Message& msg); + void messageProcessed(Message& msg); void internalListener(Consumer consumer); void receiveMessages(); void failPendingReceiveCallback(); @@ -149,8 +139,12 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase, void subscribeSingleNewConsumer(int numPartitions, TopicNamePtr topicName, int partitionIndex, ConsumerSubResultPromisePtr topicSubResultPromise, std::shared_ptr> partitionsNeedCreate); + // impl consumer base virtual method + bool hasEnoughMessagesForBatchReceive() const override; + void notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) override; private: + std::shared_ptr get_shared_this_ptr(); void setNegativeAcknowledgeEnabledForTesting(bool enabled) override; FRIEND_TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery); diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc index 5c4f7d216237a..88ba5f4056f9a 100644 --- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc +++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc @@ -4104,3 +4104,191 @@ TEST(BasicEndToEndTest, testUnAckedMessageTrackerEnabledCumulativeAck) { consumer.close(); client.close(); } + +void testBatchReceive(bool multiConsumer) { + ClientConfiguration config; + Client client(lookupUrl); + + std::string uniqueChunk = unique_str(); + std::string topicName = "persistent://public/default/test-batch-receive" + uniqueChunk; + + if (multiConsumer) { + // call admin api to make it partitioned + std::string url = + adminUrl + "admin/v2/persistent/public/default/test-batch-receive" + uniqueChunk + "/partitions"; + int res = makePutRequest(url, "5"); + LOG_INFO("res = " << res); + ASSERT_FALSE(res != 204 && res != 409); + } + + std::string subName = "subscription-name"; + Producer producer; + + Promise producerPromise; + client.createProducerAsync(topicName, WaitForCallbackValue(producerPromise)); + Future producerFuture = producerPromise.getFuture(); + Result result = producerFuture.get(producer); + ASSERT_EQ(ResultOk, result); + + Consumer consumer; + ConsumerConfiguration consumerConfig; + // when receiver queue size > maxNumMessages, use receiver queue size. + consumerConfig.setBatchReceivePolicy(BatchReceivePolicy(1000, -1, -1)); + consumerConfig.setReceiverQueueSize(10); + consumerConfig.setProperty("consumer-name", "test-consumer-name"); + consumerConfig.setProperty("consumer-id", "test-consumer-id"); + Promise consumerPromise; + client.subscribeAsync(topicName, subName, consumerConfig, + WaitForCallbackValue(consumerPromise)); + Future consumerFuture = consumerPromise.getFuture(); + result = consumerFuture.get(consumer); + ASSERT_EQ(ResultOk, result); + + // sync batch receive test + std::string prefix = "batch-receive-msg"; + int numOfMessages = 10; + for (int i = 0; i < numOfMessages; i++) { + std::string messageContent = prefix + std::to_string(i); + Message msg = MessageBuilder().setContent(messageContent).build(); + producer.send(msg); + LOG_DEBUG("1 sending message " << messageContent); + } + + Messages messages; + Result receive = consumer.batchReceive(messages); + ASSERT_EQ(receive, ResultOk); + ASSERT_EQ(messages.getMessageList().size(), numOfMessages); + + // async batch receive test + Latch latch(1); + BatchReceiveCallback batchReceiveCallback = [&latch, numOfMessages](Result result, Messages messages) { + ASSERT_EQ(result, ResultOk); + ASSERT_EQ(messages.getMessageList().size(), numOfMessages); + latch.countdown(); + }; + consumer.batchReceiveAsync(batchReceiveCallback); + for (int i = 0; i < numOfMessages; i++) { + std::string messageContent = prefix + std::to_string(i); + Message msg = MessageBuilder().setContent(messageContent).build(); + producer.send(msg); + LOG_DEBUG("2 sending message " << messageContent); + } + ASSERT_TRUE(latch.wait(std::chrono::seconds(10))); + + producer.close(); + consumer.close(); + client.close(); +} + +TEST(BasicEndToEndTest, testBatchReceive) { testBatchReceive(false); } + +TEST(BasicEndToEndTest, testBatchReceiveWithMultiConsumer) { testBatchReceive(true); } + +void testBatchReceiveTimeout(bool multiConsumer) { + ClientConfiguration config; + Client client(lookupUrl); + std::string uniqueChunk = unique_str(); + std::string topicName = "persistent://public/default/test-batch-receive-timeout" + uniqueChunk; + + if (multiConsumer) { + // call admin api to make it partitioned + std::string url = adminUrl + "admin/v2/persistent/public/default/test-batch-receive-timeout" + + uniqueChunk + "/partitions"; + int res = makePutRequest(url, "5"); + LOG_INFO("res = " << res); + ASSERT_FALSE(res != 204 && res != 409); + } + + std::string subName = "subscription-name"; + Producer producer; + + Promise producerPromise; + client.createProducerAsync(topicName, WaitForCallbackValue(producerPromise)); + Future producerFuture = producerPromise.getFuture(); + Result result = producerFuture.get(producer); + ASSERT_EQ(ResultOk, result); + + Consumer consumer; + ConsumerConfiguration consumerConfig; + consumerConfig.setBatchReceivePolicy(BatchReceivePolicy(1000, -1, 1000)); + consumerConfig.setProperty("consumer-name", "test-consumer-name"); + consumerConfig.setProperty("consumer-id", "test-consumer-id"); + Promise consumerPromise; + client.subscribeAsync(topicName, subName, consumerConfig, + WaitForCallbackValue(consumerPromise)); + Future consumerFuture = consumerPromise.getFuture(); + result = consumerFuture.get(consumer); + ASSERT_EQ(ResultOk, result); + + std::string prefix = "batch-receive-msg"; + int numOfMessages = 10; + + for (int i = 0; i < numOfMessages; i++) { + std::string messageContent = prefix + std::to_string(i); + Message msg = MessageBuilder().setContent(messageContent).build(); + producer.send(msg); + LOG_DEBUG("2 sending message " << messageContent); + } + + Latch latch(1); + BatchReceiveCallback batchReceiveCallback = [&latch, numOfMessages](Result result, Messages messages) { + ASSERT_EQ(result, ResultOk); + ASSERT_EQ(messages.getMessageList().size(), numOfMessages); + latch.countdown(); + }; + consumer.batchReceiveAsync(batchReceiveCallback); + ASSERT_TRUE(latch.wait(std::chrono::seconds(10))); + + producer.close(); + consumer.close(); + client.close(); +} + +TEST(BasicEndToEndTest, testBatchReceiveTimeout) { testBatchReceiveTimeout(false); } + +TEST(BasicEndToEndTest, testBatchReceiveTimeoutWithMultiConsumer) { testBatchReceiveTimeout(true); } + +void testBatchReceiveClose(bool multiConsumer) { + ClientConfiguration config; + Client client(lookupUrl); + + std::string uniqueChunk = unique_str(); + std::string topicName = "persistent://public/default/test-batch-receive-close" + uniqueChunk; + + if (multiConsumer) { + // call admin api to make it partitioned + std::string url = adminUrl + "admin/v2/persistent/public/default/test-batch-receive-close" + + uniqueChunk + "/partitions"; + int res = makePutRequest(url, "5"); + LOG_INFO("res = " << res); + ASSERT_FALSE(res != 204 && res != 409); + } + + std::string subName = "subscription-name"; + Consumer consumer; + ConsumerConfiguration consumerConfig; + consumerConfig.setBatchReceivePolicy(BatchReceivePolicy(1000, -1, 1000)); + consumerConfig.setProperty("consumer-name", "test-consumer-name"); + consumerConfig.setProperty("consumer-id", "test-consumer-id"); + Promise consumerPromise; + client.subscribeAsync(topicName, subName, consumerConfig, + WaitForCallbackValue(consumerPromise)); + Future consumerFuture = consumerPromise.getFuture(); + Result result = consumerFuture.get(consumer); + ASSERT_EQ(ResultOk, result); + + Latch latch(1); + BatchReceiveCallback batchReceiveCallback = [&latch](Result result, Messages messages) { + ASSERT_EQ(result, ResultAlreadyClosed); + latch.countdown(); + }; + consumer.batchReceiveAsync(batchReceiveCallback); + consumer.close(); + client.close(); + + ASSERT_TRUE(latch.wait(std::chrono::seconds(10))); +} + +TEST(BasicEndToEndTest, testBatchReceiveClose) { testBatchReceiveClose(false); } + +TEST(BasicEndToEndTest, testBatchReceiveCloseWithMultiConsumer) { testBatchReceiveClose(true); } diff --git a/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc b/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc new file mode 100644 index 0000000000000..77b2f74967575 --- /dev/null +++ b/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc @@ -0,0 +1,45 @@ +/** + * 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 +#include + +using namespace pulsar; + +TEST(BatchReceivePolicyTest, testBatchReceivePolicy) { + try { + BatchReceivePolicy batchReceivePolicy(-1, -1, -1); + FAIL() << "Should be failed."; + } catch (const std::invalid_argument& e) { + ASSERT_TRUE(true); + } + + { + BatchReceivePolicy batchReceivePolicy; + ASSERT_EQ(batchReceivePolicy.getMaxNumMessages(), -1); + ASSERT_EQ(batchReceivePolicy.getMaxNumBytes(), 10 * 1024 * 1024); + ASSERT_EQ(batchReceivePolicy.getTimeoutMs(), 100); + } + + { + BatchReceivePolicy batchReceivePolicy(-1, -1, 123); + ASSERT_EQ(batchReceivePolicy.getMaxNumMessages(), -1); + ASSERT_EQ(batchReceivePolicy.getMaxNumBytes(), 10 * 1024 * 1024); + ASSERT_EQ(batchReceivePolicy.getTimeoutMs(), 123); + } +} diff --git a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc index 24f541b57ba6a..20cd8f4bf2923 100644 --- a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc +++ b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc @@ -61,6 +61,9 @@ TEST(ConsumerConfigurationTest, testDefaultConfig) { ASSERT_EQ(conf.getPriorityLevel(), 0); ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 10); ASSERT_EQ(conf.isAutoAckOldestChunkedMessageOnQueueFull(), false); + ASSERT_EQ(conf.getBatchReceivePolicy().getMaxNumMessages(), -1); + ASSERT_EQ(conf.getBatchReceivePolicy().getMaxNumBytes(), 10 * 1024 * 1024); + ASSERT_EQ(conf.getBatchReceivePolicy().getTimeoutMs(), 100); } TEST(ConsumerConfigurationTest, testCustomConfig) { @@ -151,6 +154,11 @@ TEST(ConsumerConfigurationTest, testCustomConfig) { conf.setAutoAckOldestChunkedMessageOnQueueFull(true); ASSERT_TRUE(conf.isAutoAckOldestChunkedMessageOnQueueFull()); + + conf.setBatchReceivePolicy(BatchReceivePolicy(10, 10, 100)); + ASSERT_EQ(conf.getBatchReceivePolicy().getMaxNumMessages(), 10); + ASSERT_EQ(conf.getBatchReceivePolicy().getMaxNumBytes(), 10); + ASSERT_EQ(conf.getBatchReceivePolicy().getTimeoutMs(), 100); } TEST(ConsumerConfigurationTest, testReadCompactPersistentExclusive) { diff --git a/pulsar-client-cpp/tests/MessagesTest.cc b/pulsar-client-cpp/tests/MessagesTest.cc new file mode 100644 index 0000000000000..9cd7ea598659b --- /dev/null +++ b/pulsar-client-cpp/tests/MessagesTest.cc @@ -0,0 +1,72 @@ +/** + * 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 +#include +#include +#include "pulsar/MessageBuilder.h" + +using namespace pulsar; + +TEST(MessagesTest, testMessage) { + // 0. test not limits + { + MessagesImpl messages(-1, -1); + ASSERT_TRUE(messages.canAdd(Message())); + } + + // 1. test max number of messages. + { + Message msg = MessageBuilder().setContent("c").build(); + MessagesImpl messages(10, -1); + for (int i = 0; i < 10; i++) { + messages.add(msg); + } + ASSERT_FALSE(messages.canAdd(msg)); + ASSERT_EQ(messages.size(), 10); + try { + messages.add(msg); + FAIL() << "Should be failed."; + } catch (std::invalid_argument& e) { + } + + messages.clear(); + ASSERT_TRUE(messages.canAdd(msg)); + ASSERT_EQ(messages.size(), 0); + } + + // 2. test max size of messages. + { + Message msg = MessageBuilder().setContent("c").build(); + MessagesImpl messages(-1, 10); + for (int i = 0; i < 10; i++) { + messages.add(msg); + } + ASSERT_FALSE(messages.canAdd(msg)); + ASSERT_EQ(messages.size(), 10); + try { + messages.add(msg); + FAIL() << "Should be failed."; + } catch (std::invalid_argument& e) { + } + + messages.clear(); + ASSERT_TRUE(messages.canAdd(msg)); + ASSERT_EQ(messages.size(), 0); + } +} From 79453e0ecb5c04ca09b88bb3c7f21bf7c6cbeaf4 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Tue, 6 Sep 2022 10:39:48 +0800 Subject: [PATCH 2/6] Use weak ptr. --- pulsar-client-cpp/lib/ConsumerImplBase.cc | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImplBase.cc b/pulsar-client-cpp/lib/ConsumerImplBase.cc index 73fc75f022878..8f3d0f3a9b112 100644 --- a/pulsar-client-cpp/lib/ConsumerImplBase.cc +++ b/pulsar-client-cpp/lib/ConsumerImplBase.cc @@ -54,11 +54,12 @@ ConsumerImplBase::ConsumerImplBase(ClientImplPtr client, const std::string& topi void ConsumerImplBase::triggerBatchReceiveTimerTask(long timeoutMs) { if (timeoutMs > 0) { batchReceiveTimer_->expires_from_now(boost::posix_time::milliseconds(timeoutMs)); - auto self = shared_from_this(); - batchReceiveTimer_->async_wait([self](const boost::system::error_code& ec) { + std::weak_ptr weakSelf{shared_from_this()}; + batchReceiveTimer_->async_wait([weakSelf](const boost::system::error_code& ec) { // If two requests call runPartitionUpdateTask at the same time, the timer will fail, and it // cannot continue at this time, and the request needs to be ignored. - if (!ec) { + auto self = weakSelf.lock(); + if (self && !ec) { self->doBatchReceiveTimeTask(); } }); From 0e50258866aae668d322cc9b5bbf743633a14049 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Thu, 15 Sep 2022 10:36:34 +0800 Subject: [PATCH 3/6] Fix build failed. --- pulsar-client-cpp/lib/MessagesImpl.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-client-cpp/lib/MessagesImpl.cc b/pulsar-client-cpp/lib/MessagesImpl.cc index 15d321a484c86..7376eea3ce1ee 100644 --- a/pulsar-client-cpp/lib/MessagesImpl.cc +++ b/pulsar-client-cpp/lib/MessagesImpl.cc @@ -17,6 +17,7 @@ * under the License. */ #include "MessagesImpl.h" +#include "stdexcept" MessagesImpl::MessagesImpl(int maxNumberOfMessages, long maxSizeOfMessages) : maxNumberOfMessages_(maxNumberOfMessages), From b7c2078e202c0055df0e745ddc6622391240af64 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Wed, 21 Sep 2022 23:27:02 +0800 Subject: [PATCH 4/6] Fix code review. --- pulsar-client-cpp/.gitignore | 1 - .../include/pulsar/BatchReceivePolicy.h | 3 ++ pulsar-client-cpp/include/pulsar/Consumer.h | 1 - .../include/pulsar/ConsumerConfiguration.h | 5 +- pulsar-client-cpp/include/pulsar/Messages.h | 51 ------------------- pulsar-client-cpp/lib/ConsumerImpl.cc | 6 +-- pulsar-client-cpp/lib/Messages.cc | 34 ------------- pulsar-client-cpp/lib/MessagesImpl.h | 1 - .../lib/MultiTopicsConsumerImpl.cc | 6 +-- pulsar-client-cpp/tests/BasicEndToEndTest.cc | 6 +-- .../tests/BatchReceivePolicyTest.cc | 7 +-- .../{MessagesTest.cc => MessagesImplTest.cc} | 3 +- 12 files changed, 15 insertions(+), 109 deletions(-) delete mode 100644 pulsar-client-cpp/include/pulsar/Messages.h delete mode 100644 pulsar-client-cpp/lib/Messages.cc rename pulsar-client-cpp/tests/{MessagesTest.cc => MessagesImplTest.cc} (97%) diff --git a/pulsar-client-cpp/.gitignore b/pulsar-client-cpp/.gitignore index 9111dd42202f5..8c8c065e61935 100644 --- a/pulsar-client-cpp/.gitignore +++ b/pulsar-client-cpp/.gitignore @@ -77,7 +77,6 @@ Makefile cmake_install.cmake CMakeFiles CMakeCache.txt -build/ pulsar-dist install_manifest.txt diff --git a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h index ca9c4046b40fc..489d93a5d9f4b 100644 --- a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h +++ b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h @@ -50,6 +50,9 @@ struct BatchReceivePolicyImpl; */ class PULSAR_PUBLIC BatchReceivePolicy { public: + /** + * Default value: {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100} + */ BatchReceivePolicy(); /** diff --git a/pulsar-client-cpp/include/pulsar/Consumer.h b/pulsar-client-cpp/include/pulsar/Consumer.h index 907d26af93f12..b702cff771a06 100644 --- a/pulsar-client-cpp/include/pulsar/Consumer.h +++ b/pulsar-client-cpp/include/pulsar/Consumer.h @@ -23,7 +23,6 @@ #include #include #include -#include namespace pulsar { class PulsarWrapper; diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index c6fff393754ab..8da29e10c8366 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -31,7 +31,6 @@ #include #include #include -#include #include "BatchReceivePolicy.h" namespace pulsar { @@ -40,6 +39,7 @@ class Consumer; class PulsarWrapper; /// Callback definition for non-data operation +typedef std::vector Messages; typedef std::function ResultCallback; typedef std::function ReceiveCallback; typedef std::function BatchReceiveCallback; @@ -384,7 +384,8 @@ class PULSAR_PUBLIC ConsumerConfiguration { /** * Set batch receive policy. * - * @param batchReceivePolicy the default is xxx + * @param batchReceivePolicy the default is + * {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100} */ void setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy); diff --git a/pulsar-client-cpp/include/pulsar/Messages.h b/pulsar-client-cpp/include/pulsar/Messages.h deleted file mode 100644 index f0157f057ac1d..0000000000000 --- a/pulsar-client-cpp/include/pulsar/Messages.h +++ /dev/null @@ -1,51 +0,0 @@ -/** - * 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. - */ -#ifndef MESSAGES_HPP_ -#define MESSAGES_HPP_ - -#include -#include -#include - -namespace pulsar { - -class Message; -class MessagesImpl; - -class PULSAR_PUBLIC Messages { - public: - Messages(); - - /** - * Get message list. - * - * @return message list. - */ - std::vector getMessageList() const; - - private: - typedef std::shared_ptr MessagesImplPtr; - MessagesImplPtr impl_; - Messages(MessagesImplPtr msgsPtr); - friend class ConsumerImpl; - friend class MultiTopicsConsumerImpl; -}; -} // namespace pulsar - -#endif /* MESSAGES_HPP_ */ diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index c68fec7fb0077..50301f15e915f 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -529,10 +529,8 @@ void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback messages->add(msg); } auto self = get_shared_this_ptr(); - listenerExecutor_->postWork([callback, messages, self]() { - Messages msgs(messages); - callback(ResultOk, msgs); - }); + listenerExecutor_->postWork( + [callback, messages, self]() { callback(ResultOk, messages->getMessageList()); }); } void ConsumerImpl::notifyPendingReceivedCallback(Result result, Message& msg, diff --git a/pulsar-client-cpp/lib/Messages.cc b/pulsar-client-cpp/lib/Messages.cc deleted file mode 100644 index fee2ba5e2d479..0000000000000 --- a/pulsar-client-cpp/lib/Messages.cc +++ /dev/null @@ -1,34 +0,0 @@ -/** - * 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 -#include -#include "MessagesImpl.h" - -using namespace pulsar; - -namespace pulsar { - -Messages::Messages() : impl_() {} - -std::vector Messages::getMessageList() const { return impl_->getMessageList(); } - -Messages::Messages(Messages::MessagesImplPtr msgsPtr) : impl_(msgsPtr) {} - -} // namespace pulsar diff --git a/pulsar-client-cpp/lib/MessagesImpl.h b/pulsar-client-cpp/lib/MessagesImpl.h index 1c4d556c6d76d..feb866e9df47f 100644 --- a/pulsar-client-cpp/lib/MessagesImpl.h +++ b/pulsar-client-cpp/lib/MessagesImpl.h @@ -21,7 +21,6 @@ #include #include -#include using namespace pulsar; diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc index 5c7bb929aa85d..6137a1a8687eb 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc @@ -873,10 +873,8 @@ void MultiTopicsConsumerImpl::notifyBatchPendingReceivedCallback(const BatchRece messages->add(msg); } auto self = get_shared_this_ptr(); - listenerExecutor_->postWork([callback, messages, self]() { - Messages msgs(messages); - callback(ResultOk, msgs); - }); + listenerExecutor_->postWork( + [callback, messages, self]() { callback(ResultOk, messages->getMessageList()); }); } void MultiTopicsConsumerImpl::messageProcessed(Message& msg) { diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc index 88ba5f4056f9a..d5163d07fa3fa 100644 --- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc +++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc @@ -4157,13 +4157,13 @@ void testBatchReceive(bool multiConsumer) { Messages messages; Result receive = consumer.batchReceive(messages); ASSERT_EQ(receive, ResultOk); - ASSERT_EQ(messages.getMessageList().size(), numOfMessages); + ASSERT_EQ(messages.size(), numOfMessages); // async batch receive test Latch latch(1); BatchReceiveCallback batchReceiveCallback = [&latch, numOfMessages](Result result, Messages messages) { ASSERT_EQ(result, ResultOk); - ASSERT_EQ(messages.getMessageList().size(), numOfMessages); + ASSERT_EQ(messages.size(), numOfMessages); latch.countdown(); }; consumer.batchReceiveAsync(batchReceiveCallback); @@ -4233,7 +4233,7 @@ void testBatchReceiveTimeout(bool multiConsumer) { Latch latch(1); BatchReceiveCallback batchReceiveCallback = [&latch, numOfMessages](Result result, Messages messages) { ASSERT_EQ(result, ResultOk); - ASSERT_EQ(messages.getMessageList().size(), numOfMessages); + ASSERT_EQ(messages.size(), numOfMessages); latch.countdown(); }; consumer.batchReceiveAsync(batchReceiveCallback); diff --git a/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc b/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc index 77b2f74967575..ab9ffcc0ff252 100644 --- a/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc +++ b/pulsar-client-cpp/tests/BatchReceivePolicyTest.cc @@ -22,12 +22,7 @@ using namespace pulsar; TEST(BatchReceivePolicyTest, testBatchReceivePolicy) { - try { - BatchReceivePolicy batchReceivePolicy(-1, -1, -1); - FAIL() << "Should be failed."; - } catch (const std::invalid_argument& e) { - ASSERT_TRUE(true); - } + ASSERT_THROW(BatchReceivePolicy(-1, -1, -1), std::invalid_argument); { BatchReceivePolicy batchReceivePolicy; diff --git a/pulsar-client-cpp/tests/MessagesTest.cc b/pulsar-client-cpp/tests/MessagesImplTest.cc similarity index 97% rename from pulsar-client-cpp/tests/MessagesTest.cc rename to pulsar-client-cpp/tests/MessagesImplTest.cc index 9cd7ea598659b..e042fc12d1913 100644 --- a/pulsar-client-cpp/tests/MessagesTest.cc +++ b/pulsar-client-cpp/tests/MessagesImplTest.cc @@ -17,13 +17,12 @@ * under the License. */ #include -#include #include #include "pulsar/MessageBuilder.h" using namespace pulsar; -TEST(MessagesTest, testMessage) { +TEST(MessagesImplTest, testMessage) { // 0. test not limits { MessagesImpl messages(-1, -1); From 77ed37e096f1b94ab6fe5eea8c7291970cac4e5d Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Thu, 22 Sep 2022 10:41:58 +0800 Subject: [PATCH 5/6] Fix docs. --- pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h | 4 ++-- pulsar-client-cpp/include/pulsar/Consumer.h | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h index 489d93a5d9f4b..844eade35c883 100644 --- a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h +++ b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h @@ -34,13 +34,13 @@ struct BatchReceivePolicyImpl; * timeout for waiting for enough messages for this batch. * *

This batch receive will be completed as long as any one of the - * conditions(has enough number of messages, has enough of size of messages, wait timeout) is met. + * conditions(has enough number of messages, has enough size of messages, wait timeout) are met. * *

Examples: * 1.If set maxNumMessages = 10, maxSizeOfMessages = 1MB and without timeout, it * means {@link Consumer#batchReceive()} will always wait until there is enough messages. * 2.If set maxNumberOfMessages = 0, maxNumBytes = 0 and timeout = 100ms, it - * means {@link Consumer#batchReceive()} will waiting for 100ms whether or not there is enough messages. + * means {@link Consumer#batchReceive()} will wait for 100ms no matter whether there are enough messages. * *

Note: * Must specify messages limitation(maxNumMessages, maxNumBytes) or wait timeout. diff --git a/pulsar-client-cpp/include/pulsar/Consumer.h b/pulsar-client-cpp/include/pulsar/Consumer.h index b702cff771a06..c7911b982461e 100644 --- a/pulsar-client-cpp/include/pulsar/Consumer.h +++ b/pulsar-client-cpp/include/pulsar/Consumer.h @@ -134,7 +134,7 @@ class PULSAR_PUBLIC Consumer { * batchReceiveAsync() should be called subsequently once callback gets completed with received message. * Else it creates backlog of receive requests in the application. *

- * @param BatchReceiveCallback will be completed when messages is available + * @param BatchReceiveCallback will be completed when messages are available. */ void batchReceiveAsync(BatchReceiveCallback callback); From 02a66d52c6e588424068250ee5c97d5d4dba448c Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Thu, 22 Sep 2022 15:46:08 +0800 Subject: [PATCH 6/6] Fix docs. --- pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h index 844eade35c883..3c66da2f7e5c2 100644 --- a/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h +++ b/pulsar-client-cpp/include/pulsar/BatchReceivePolicy.h @@ -33,8 +33,8 @@ struct BatchReceivePolicyImpl; *

Batch receive policy can limit the number and bytes of messages in a single batch, and can specify a * timeout for waiting for enough messages for this batch. * - *

This batch receive will be completed as long as any one of the - * conditions(has enough number of messages, has enough size of messages, wait timeout) are met. + *

A batch receive action is completed as long as any one of the + * conditions (the batch has enough number or size of messages, or the waiting timeout is passed) are met. * *

Examples: * 1.If set maxNumMessages = 10, maxSizeOfMessages = 1MB and without timeout, it @@ -71,7 +71,7 @@ class PULSAR_PUBLIC BatchReceivePolicy { long getTimeoutMs() const; /** - * Get max num messages. + * Get the maximum number of messages. * @return */ int getMaxNumMessages() const;