[feat] Consumer support batch receive messages. (#21)

### Motivation
apache/pulsar#17140

This PR has been reviewed in [pulsar repo](apache/pulsar#17429).

### Modifications
- Consumer support batch receives messages.
- Abstract common implementation to `ConsumerImplBase`.

Author: shibd

include/pulsar/BatchReceivePolicy.h

@@ -0,0 +1,90 @@
+/**
+ * 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 <pulsar/defines.h>
+#include <memory>
+
+namespace pulsar {
+
+struct BatchReceivePolicyImpl;
+
+/**
+ * Configuration for message batch receive {@link Consumer#batchReceive()} {@link
+ * Consumer#batchReceiveAsync()}.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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 wait for 100ms no matter whether there are enough messages.
+ *
+ * <p>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:
+    /**
+     * Default value: {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100}
+     */
+    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 the maximum number of messages.
+     * @return
+     */
+    int getMaxNumMessages() const;
+
+    /**
+     * Get max num bytes.
+     * @return
+     */
+    long getMaxNumBytes() const;
+
+   private:
+    std::shared_ptr<BatchReceivePolicyImpl> impl_;
+};
+}  // namespace pulsar
+
+#endif /* BATCH_RECEIVE_POLICY_HPP_ */

include/pulsar/Consumer.h

@@ -113,6 +113,31 @@ class PULSAR_PUBLIC Consumer {
      */
     void receiveAsync(ReceiveCallback callback);
 
+    /**
+     * Batch receiving messages.
+     *
+     * <p>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.
+     * <p>
+     * Retrieves a message when it will be available and completes callback with received message.
+     * </p>
+     * <p>
+     * batchReceiveAsync() should be called subsequently once callback gets completed with received message.
+     * Else it creates <i> backlog of receive requests </i> in the application.
+     * </p>
+     * @param BatchReceiveCallback will be completed when messages are available.
+     */
+    void batchReceiveAsync(BatchReceiveCallback callback);
+
     /**
      * Acknowledge the reception of a single message.
      *

include/pulsar/ConsumerConfiguration.h

@@ -31,15 +31,18 @@
 #include <pulsar/InitialPosition.h>
 #include <pulsar/KeySharedPolicy.h>
 #include <pulsar/ConsumerEventListener.h>
+#include "BatchReceivePolicy.h"
 
 namespace pulsar {
 
 class Consumer;
 class PulsarWrapper;
 
 /// Callback definition for non-data operation
+typedef std::vector<Message> Messages;
 typedef std::function<void(Result result)> ResultCallback;
 typedef std::function<void(Result, const Message& msg)> ReceiveCallback;
+typedef std::function<void(Result, const Messages& msgs)> BatchReceiveCallback;
 typedef std::function<void(Result result, MessageId messageId)> GetLastMessageIdCallback;
 
 /// Callback definition for MessageListener
@@ -378,6 +381,21 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     InitialPosition getSubscriptionInitialPosition() const;
 
+    /**
+     * Set batch receive policy.
+     *
+     * @param batchReceivePolicy the default is
+     * {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100}
+     */
+    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`.

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 <pulsar/BatchReceivePolicy.h>
+#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<BatchReceivePolicyImpl>()) {
+    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

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.
+ */
+#pragma once
+
+namespace pulsar {
+
+struct BatchReceivePolicyImpl {
+    int maxNumMessage;
+    long maxNumBytes;
+    long timeoutMs;
+};
+
+}  // namespace pulsar

lib/ClientImpl.h

@@ -28,14 +28,12 @@
 #include <mutex>
 #include <lib/TopicName.h>
 #include "ProducerImplBase.h"
-#include "ConsumerImplBase.h"
 #include <atomic>
 #include <vector>
 #include "ServiceNameResolver.h"
 
 namespace pulsar {
 
-class ClientImpl;
 class PulsarFriend;
 typedef std::shared_ptr<ClientImpl> ClientImplPtr;
 typedef std::weak_ptr<ClientImpl> ClientImplWeakPtr;
@@ -44,6 +42,9 @@ class ReaderImpl;
 typedef std::shared_ptr<ReaderImpl> ReaderImplPtr;
 typedef std::weak_ptr<ReaderImpl> ReaderImplWeakPtr;
 
+class ConsumerImplBase;
+typedef std::weak_ptr<ConsumerImplBase> ConsumerImplBaseWeakPtr;
+
 std::string generateRandomName();
 
 class ClientImpl : public std::enable_shared_from_this<ClientImpl> {

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<Result, Messages> promise;
+    impl_->batchReceiveAsync(WaitForCallbackValue<Messages>(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) {

lib/ConsumerConfiguration.cc

@@ -19,6 +19,7 @@
 #include <lib/ConsumerConfigurationImpl.h>
 
 #include <stdexcept>
+#include <pulsar/ConsumerConfiguration.h>
 
 namespace pulsar {
 
@@ -267,4 +268,12 @@ ConsumerConfiguration& ConsumerConfiguration::setStartMessageIdInclusive(bool st
 
 bool ConsumerConfiguration::isStartMessageIdInclusive() const { return impl_->startMessageIdInclusive; }
 
+void ConsumerConfiguration::setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy) {
+    impl_->batchReceivePolicy = batchReceivePolicy;
+}
+
+const BatchReceivePolicy& ConsumerConfiguration::getBatchReceivePolicy() const {
+    return impl_->batchReceivePolicy;
+}
+
 }  // namespace pulsar

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<std::string, std::string> properties;