[feat][cpp] Consumer support batch receive messages.

.asf.yaml

@@ -43,13 +43,18 @@ github:
     rebase:  false
   protected_branches:
     main:
-      # required_status_checks:
+      required_status_checks:
       #   # strict means "Require branches to be up to date before merging".
-      #   strict: false
+        strict: false
       #   # Contexts are the names of checks that must pass.
       #   # See ./github/workflows/README.md for more documentation on this list.
-      #   contexts:
-      #      - Pulsar CI checks completed
+        contexts:
+           - Run unit tests
+           - Build CPP Client on Windows x64
+           - Build CPP Client on Windows x86
+           - Build Debian Package
+           - Build RPM Package
+           - Build Alpine Linux APK Package
 
       required_pull_request_reviews:
         dismiss_stale_reviews: false

CMakeLists.txt

@@ -122,36 +122,6 @@ endif(NOT LOG_CATEGORY_NAME)
 
 add_definitions(-DLOG_CATEGORY_NAME=${LOG_CATEGORY_NAME} -DBUILDING_PULSAR -DBOOST_ALL_NO_LIB -DBOOST_ALLOW_DEPRECATED_HEADERS)
 
-set(OPENSSL_ROOT_DIR ${OPENSSL_ROOT_DIR} /usr/lib64/)
-
-### This part is to find and keep SSL dynamic libs in RECORD_OPENSSL_SSL_LIBRARY and RECORD_OPENSSL_CRYPTO_LIBRARY
-### After find the libs, will unset related cache, and will not affect another same call to find_package.
-if (APPLE)
-    set(OPENSSL_INCLUDE_DIR /usr/local/opt/openssl/include/ /opt/homebrew/opt/openssl/include)
-    set(OPENSSL_ROOT_DIR ${OPENSSL_ROOT_DIR} /usr/local/opt/openssl/ /opt/homebrew/opt/openssl)
-endif ()
-
-set(OPENSSL_USE_STATIC_LIBS FALSE)
-find_package(OpenSSL REQUIRED)
-set(RECORD_OPENSSL_SSL_LIBRARY ${OPENSSL_SSL_LIBRARY})
-set(RECORD_OPENSSL_CRYPTO_LIBRARY ${OPENSSL_CRYPTO_LIBRARY})
-message("RECORD_OPENSSL_SSL_LIBRARY: " ${RECORD_OPENSSL_SSL_LIBRARY})
-message("RECORD_OPENSSL_CRYPTO_LIBRARY: " ${RECORD_OPENSSL_CRYPTO_LIBRARY})
-
-unset(OPENSSL_FOUND CACHE)
-unset(OPENSSL_INCLUDE_DIR CACHE)
-unset(OPENSSL_CRYPTO_LIBRARY CACHE)
-unset(OPENSSL_CRYPTO_LIBRARIES CACHE)
-unset(OPENSSL_SSL_LIBRARY CACHE)
-unset(OPENSSL_SSL_LIBRARIES CACHE)
-unset(OPENSSL_LIBRARIES CACHE)
-unset(OPENSSL_VERSION CACHE)
-
-find_package(OpenSSL REQUIRED)
-message("OPENSSL_INCLUDE_DIR: " ${OPENSSL_INCLUDE_DIR})
-message("OPENSSL_SSL_LIBRARY: " ${OPENSSL_SSL_LIBRARY})
-message("OPENSSL_CRYPTO_LIBRARY: " ${OPENSSL_CRYPTO_LIBRARY})
-
 # For dependencies other than OpenSSL, dynamic libraries are forbidden to link when LINK_STATIC is ON
 if (LINK_STATIC)
     if (NOT MSVC)
@@ -162,6 +132,14 @@ endif ()
 find_package(Boost REQUIRED)
 message("Boost_INCLUDE_DIRS: " ${Boost_INCLUDE_DIRS})
 
+set(OPENSSL_ROOT_DIR ${OPENSSL_ROOT_DIR} /usr/lib64/)
+if (APPLE)
+    set(OPENSSL_ROOT_DIR ${OPENSSL_ROOT_DIR} /usr/local/opt/openssl/ /opt/homebrew/opt/openssl)
+endif ()
+find_package(OpenSSL REQUIRED)
+message("OPENSSL_INCLUDE_DIR: " ${OPENSSL_INCLUDE_DIR})
+message("OPENSSL_LIBRARIES: " ${OPENSSL_LIBRARIES})
+
 find_package(Protobuf REQUIRED)
 message("Protobuf_INCLUDE_DIRS: " ${Protobuf_INCLUDE_DIRS})
 message("Protobuf_LIBRARIES: " ${Protobuf_LIBRARIES})
@@ -217,9 +195,6 @@ if (LINK_STATIC)
     if (MSVC)
         add_definitions(-DCURL_STATICLIB)
     endif()
-
-    SET(Boost_USE_STATIC_LIBS   ON)
-    SET(OPENSSL_USE_STATIC_LIBS TRUE)
 else()
     if (MSVC AND (${CMAKE_BUILD_TYPE} STREQUAL Debug))
         find_library(LIB_ZSTD zstdd HINTS "${VCPKG_DEBUG_ROOT}/lib")

README.md

@@ -92,9 +92,9 @@ Run unit tests:
 #### Install all dependencies:
 
 ```shell
-apt-get install -y g++ cmake libssl-dev libcurl4-openssl-dev liblog4cxx-dev \
-                libprotobuf-dev libboost-all-dev  libgtest-dev google-mock \
-                protobuf-compiler python3-setuptools
+apt-get install -y g++ cmake libssl-dev libcurl4-openssl-dev \
+                libprotobuf-dev libboost-all-dev libgtest-dev google-mock \
+                protobuf-compiler
 ```
 
 #### Compile and install Google Test:
@@ -105,8 +105,7 @@ sudo cmake .
 sudo make
 
 # Copy the libraries you just built to the OS library path.
-# GTEST_LIB_PATH may be `/usr/src/gtest`, `/usr/src/gtest/lib` or other path you provided when building gtest above.
-sudo cp ${GTEST_LIB_PATH}/*.a /usr/lib
+sudo cp lib/*.a /usr/lib
 ```
 
 
@@ -117,9 +116,10 @@ cd /usr/src/gmock
 sudo cmake .
 sudo make
 
-# Copy the libraries you just built to the OS library path.
-# GMOCK_LIB_PATH may be `/usr/src/gmock`, `/usr/src/gmock/lib` or other path you provided when building gmock above.
-sudo cp ${GMOCK_LIB_PATH}/*.a /usr/lib
+# Copy the gmock headers to the OS include path.
+sudo cp -r include/gmock /usr/include/
+# Copy the libraries you just built to the OS brary path.
+sudo cp lib/*.a /usr/lib
 ```
 
 

docker-build.sh

@@ -24,7 +24,7 @@
 set -e
 
 ROOT_DIR=$(git rev-parse --show-toplevel)
-cd $ROOT_DIR/pulsar-client-cpp
+cd $ROOT_DIR
 
 BUILD_IMAGE_NAME="${BUILD_IMAGE_NAME:-apachepulsar/pulsar-build}"
 BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-20.04}"
@@ -35,8 +35,8 @@ echo "---- Build Pulsar C++ client using image $IMAGE (pass <skip-clean> for inc
 
 docker pull $IMAGE
 
-VOLUME_OPTION=${VOLUME_OPTION:-"-v $ROOT_DIR:/pulsar"}
-COMMAND="cd /pulsar/pulsar-client-cpp && cmake . $CMAKE_ARGS && make check-format && make -j8"
+VOLUME_OPTION=${VOLUME_OPTION:-"-v $ROOT_DIR:/pulsar-client-cpp"}
+COMMAND="cd /pulsar-client-cpp && cmake . $CMAKE_ARGS && make check-format && make -j8"
 
 DOCKER_CMD="docker run -i ${VOLUME_OPTION} ${IMAGE}"
 

docker-format.sh

@@ -24,7 +24,7 @@
 set -e
 
 ROOT_DIR=$(git rev-parse --show-toplevel)
-cd $ROOT_DIR/pulsar-client-cpp
+cd $ROOT_DIR
 
 BUILD_IMAGE_NAME="${BUILD_IMAGE_NAME:-apachepulsar/pulsar-build}"
 BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-20.04}"
@@ -35,8 +35,8 @@ echo "---- Build Pulsar C++ client using image $IMAGE"
 
 docker pull $IMAGE
 
-VOLUME_OPTION=${VOLUME_OPTION:-"-v $ROOT_DIR:/pulsar"}
-COMMAND="cd /pulsar/pulsar-client-cpp && cmake . $CMAKE_ARGS && make format"
+VOLUME_OPTION=${VOLUME_OPTION:-"-v $ROOT_DIR:/pulsar-client-cpp"}
+COMMAND="cd /pulsar-client-cpp && cmake . $CMAKE_ARGS && make format"
 
 DOCKER_CMD="docker run -i ${VOLUME_OPTION} ${IMAGE}"
 

docker-tests.sh

@@ -31,7 +31,7 @@ fi
 
 
 ROOT_DIR=$(git rev-parse --show-toplevel)
-cd $ROOT_DIR/pulsar-client-cpp
+cd $ROOT_DIR
 
 BUILD_IMAGE_NAME="${BUILD_IMAGE_NAME:-apachepulsar/pulsar-build}"
 BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-20.04}"
@@ -44,7 +44,7 @@ docker pull $IMAGE
 
 CONTAINER_LABEL="pulsartests=$$"
 export GTEST_COLOR=${GTEST_COLOR:-no}
-DOCKER_CMD="docker run -e GTEST_COLOR -i -l $CONTAINER_LABEL -v $ROOT_DIR:/pulsar $IMAGE"
+DOCKER_CMD="docker run -e GTEST_COLOR -i -l $CONTAINER_LABEL -v $ROOT_DIR:/pulsar-client-cpp $IMAGE"
 
 
 for args in "$@"
@@ -68,7 +68,7 @@ fi
 
 # Java17 is required for CLI e.g) bin/pulsar create-token
 $DOCKER_CMD bash -c "apt-get -y install openjdk-17-jre-headless &&\
- set -o pipefail; cd /pulsar/pulsar-client-cpp && ./run-unit-tests.sh ${tests} $DISABLE_COLOR_OUTPUT"
+ set -o pipefail; cd /pulsar-client-cpp && git config --global --add safe.directory /pulsar-client-cpp && ./run-unit-tests.sh ${tests} $DISABLE_COLOR_OUTPUT"
 RES=$?
 if [ $RES -ne 0 ]; then
   (

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`.