Merge main

Author: RobertIndie

.github/workflows/ci-pr-validation.yaml

@@ -29,6 +29,34 @@ concurrency:
 
 jobs:
 
+  wireshark-dissector-build:
+    name: Build the Wireshark dissector
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 60
+    strategy:
+      matrix:
+        os: [ubuntu-20.04, macos-12]
+
+    steps:
+      - name: checkout
+        uses: actions/checkout@v3
+
+      - name: Install deps (Ubuntu)
+        if: ${{ startsWith(matrix.os, 'ubuntu') }}
+        run: |
+          sudo apt-get update -y
+          sudo apt-get install -y protobuf-compiler libprotobuf-dev wireshark-dev
+
+      - name: Install deps (macOS)
+        if: ${{ startsWith(matrix.os, 'macos') }}
+        run:
+          brew install wireshark protobuf
+
+      - name: Build wireshark plugin
+        run: |
+          cmake -S wireshark -B build-wireshark
+          cmake --build build-wireshark
+
   unit-tests:
     name: Run unit tests
     runs-on: ubuntu-22.04
@@ -264,7 +292,7 @@ jobs:
   check-completion:
     name: Check Completion
     runs-on: ubuntu-latest
-    needs: [unit-tests, cpp-build-windows, package, cpp-build-macos]
+    needs: [wireshark-dissector-build, unit-tests, cpp-build-windows, package, cpp-build-macos]
 
     steps:
       - run: true

CMakeLists.txt

@@ -71,9 +71,6 @@ MESSAGE(STATUS "BUILD_STATIC_LIB:  " ${BUILD_STATIC_LIB})
 option(BUILD_TESTS "Build tests" ON)
 MESSAGE(STATUS "BUILD_TESTS:  " ${BUILD_TESTS})
 
-option(BUILD_WIRESHARK "Build Pulsar Wireshark dissector" OFF)
-MESSAGE(STATUS "BUILD_WIRESHARK:  " ${BUILD_WIRESHARK})
-
 option(BUILD_PERF_TOOLS "Build Pulsar CLI perf producer/consumer" OFF)
 MESSAGE(STATUS "BUILD_PERF_TOOLS:  " ${BUILD_PERF_TOOLS})
 
@@ -416,10 +413,6 @@ if (BUILD_TESTS)
     add_subdirectory(tests)
 endif()
 
-if (BUILD_WIRESHARK)
-    add_subdirectory(wireshark)
-endif()
-
 find_package(ClangTools)
 set(BUILD_SUPPORT_DIR "${PROJECT_SOURCE_DIR}/build-support")
 add_custom_target(format ${BUILD_SUPPORT_DIR}/run_clang_format.py

README.md

@@ -225,6 +225,10 @@ cd tests
 ./pulsar-test-service-stop.sh
 ```
 
+## Wireshark Dissector
+
+See the [wireshark](wireshark/) directory for details.
+
 ## Requirements for Contributors
 
 It's required to install [LLVM](https://llvm.org/builds/) for `clang-tidy` and `clang-format`. Pulsar C++ client use `clang-format` **11** to format files.  `make format` automatically formats the files.

build-support/setup-test-service-container.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+set -e -x
+
+if [ $# -ne 2 ]; then
+    echo "Usage: $0 \$CONTAINER_ID \$START_TEST_SERVICE_INSIDE_CONTAINER"
+    exit 1
+fi
+
+CONTAINER_ID=$1
+START_TEST_SERVICE_INSIDE_CONTAINER=$2
+
+echo $CONTAINER_ID >> .tests-container-id.txt
+
+docker cp test-conf $CONTAINER_ID:/pulsar/test-conf
+docker cp build-support/$START_TEST_SERVICE_INSIDE_CONTAINER $CONTAINER_ID:$START_TEST_SERVICE_INSIDE_CONTAINER
+
+docker exec -i $CONTAINER_ID /$START_TEST_SERVICE_INSIDE_CONTAINER

build-support/start-mim-test-service-inside-container.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+set -e -x
+
+export PULSAR_EXTRA_OPTS=-Dpulsar.auth.basic.conf=test-conf/.htpasswd
+
+# Generate secret key and token
+mkdir -p data/tokens
+bin/pulsar tokens create-secret-key --output data/tokens/secret.key
+
+bin/pulsar tokens create \
+            --subject token-principal \
+            --secret-key file:///pulsar/data/tokens/secret.key \
+            > /pulsar/data/tokens/token.txt
+
+export PULSAR_STANDALONE_CONF=test-conf/standalone-ssl-mim.conf
+export PULSAR_PID_DIR=/tmp
+bin/pulsar-daemon start standalone \
+        --no-functions-worker --no-stream-storage \
+        --bookkeeper-dir data/bookkeeper
+
+echo "-- Wait for Pulsar service to be ready"
+until curl http://localhost:8081/metrics > /dev/null 2>&1 ; do sleep 1; done
+
+echo "-- Pulsar service is ready -- Configure permissions"
+
+export PULSAR_CLIENT_CONF=test-conf/client-ssl-mim.conf
+
+# Create "standalone" cluster if it does not exist
+bin/pulsar-admin clusters list | grep -q '^standalone$' ||
+        bin/pulsar-admin clusters create \
+                standalone \
+                --url http://localhost:8081/ \
+                --url-secure https://localhost:8444/ \
+                --broker-url pulsar://localhost:6652/ \
+                --broker-url-secure pulsar+ssl://localhost:6653/
+
+# Create "private" tenant
+bin/pulsar-admin tenants create private -r "" -c "standalone"
+
+# Create "private/auth" with required authentication
+bin/pulsar-admin namespaces create private/auth --clusters standalone
+
+bin/pulsar-admin namespaces grant-permission private/auth \
+                        --actions produce,consume \
+                        --role "token-principal"
+
+echo "-- Ready to start tests"

include/pulsar/Consumer.h

@@ -21,6 +21,7 @@
 
 #include <pulsar/BrokerConsumerStats.h>
 #include <pulsar/ConsumerConfiguration.h>
+#include <pulsar/TypedMessage.h>
 #include <pulsar/defines.h>
 
 #include <iostream>
@@ -91,6 +92,14 @@ class PULSAR_PUBLIC Consumer {
      */
     Result receive(Message& msg);
 
+    template <typename T>
+    Result receive(TypedMessage<T>& msg, typename TypedMessage<T>::Decoder decoder) {
+        Message rawMsg;
+        auto result = receive(rawMsg);
+        msg = TypedMessage<T>{rawMsg, decoder};
+        return result;
+    }
+
     /**
      *
      * @param msg a non-const reference where the received message will be copied
@@ -101,6 +110,14 @@ class PULSAR_PUBLIC Consumer {
      */
     Result receive(Message& msg, int timeoutMs);
 
+    template <typename T>
+    Result receive(TypedMessage<T>& msg, int timeoutMs, typename TypedMessage<T>::Decoder decoder) {
+        Message rawMsg;
+        auto result = receive(rawMsg, timeoutMs);
+        msg = TypedMessage<T>{rawMsg, decoder};
+        return result;
+    }
+
     /**
      * Receive a single message
      * <p>
@@ -114,6 +131,14 @@ class PULSAR_PUBLIC Consumer {
      */
     void receiveAsync(ReceiveCallback callback);
 
+    template <typename T>
+    void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
+                      typename TypedMessage<T>::Decoder decoder) {
+        receiveAsync([callback, decoder](Result result, const Message& msg) {
+            callback(result, TypedMessage<T>{msg, decoder});
+        });
+    }
+
     /**
      * Batch receiving messages.
      *

include/pulsar/ConsumerConfiguration.h

@@ -28,17 +28,20 @@
 #include <pulsar/Message.h>
 #include <pulsar/Result.h>
 #include <pulsar/Schema.h>
+#include <pulsar/TypedMessage.h>
 #include <pulsar/defines.h>
 
 #include <functional>
 #include <memory>
 
 #include "BatchReceivePolicy.h"
+#include "DeadLetterPolicy.h"
 
 namespace pulsar {
 
 class Consumer;
 class PulsarWrapper;
+class PulsarFriend;
 
 /// Callback definition for non-data operation
 typedef std::vector<Message> Messages;
@@ -48,7 +51,7 @@ typedef std::function<void(Result, const Messages& msgs)> BatchReceiveCallback;
 typedef std::function<void(Result result, MessageId messageId)> GetLastMessageIdCallback;
 
 /// Callback definition for MessageListener
-typedef std::function<void(Consumer consumer, const Message& msg)> MessageListener;
+typedef std::function<void(Consumer& consumer, const Message& msg)> MessageListener;
 
 typedef std::shared_ptr<ConsumerEventListener> ConsumerEventListenerPtr;
 
@@ -126,6 +129,15 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     ConsumerConfiguration& setMessageListener(MessageListener messageListener);
 
+    template <typename T>
+    ConsumerConfiguration& setTypedMessageListener(
+        std::function<void(Consumer&, const TypedMessage<T>&)> listener,
+        typename TypedMessage<T>::Decoder decoder) {
+        return setMessageListener([listener, decoder](Consumer& consumer, const Message& msg) {
+            listener(consumer, TypedMessage<T>{msg, decoder});
+        });
+    }
+
     /**
      * @return the message listener
      */
@@ -398,6 +410,42 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     const BatchReceivePolicy& getBatchReceivePolicy() const;
 
+    /**
+     * Set dead letter policy for consumer
+     *
+     * By default, some messages are redelivered many times, even to the extent that they can never be
+     * stopped. By using the dead letter mechanism, messages have the max redelivery count, when they
+     * exceeding the maximum number of redeliveries. Messages are sent to dead letter topics and acknowledged
+     * automatically.
+     *
+     * You can enable the dead letter mechanism by setting the dead letter policy.
+     * Example:
+     *
+     * <pre>
+     * * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder()
+     *                       .maxRedeliverCount(10)
+     *                       .build();
+     * </pre>
+     * Default dead letter topic name is {TopicName}-{Subscription}-DLQ.
+     * To set a custom dead letter topic name
+     * <pre>
+     * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder()
+     *                       .deadLetterTopic("dlq-topic")
+     *                       .maxRedeliverCount(10)
+     *                       .initialSubscriptionName("init-sub-name")
+     *                       .build();
+     * </pre>
+     * @param deadLetterPolicy Default value is empty
+     */
+    void setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy);
+
+    /**
+     * Get dead letter policy.
+     *
+     * @return dead letter policy
+     */
+    const DeadLetterPolicy& getDeadLetterPolicy() const;
+
     /**
      * Set whether the subscription status should be replicated.
      * The default value is `false`.
@@ -553,7 +601,25 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     bool isStartMessageIdInclusive() const;
 
+    /**
+     * Enable the batch index acknowledgment.
+     *
+     * It should be noted that this option can only work when the broker side also enables the batch index
+     * acknowledgment. See the `acknowledgmentAtBatchIndexLevelEnabled` config in `broker.conf`.
+     *
+     * Default: false
+     *
+     * @param enabled whether to enable the batch index acknowledgment
+     */
+    ConsumerConfiguration& setBatchIndexAckEnabled(bool enabled);
+
+    /**
+     * The associated getter of setBatchingEnabled
+     */
+    bool isBatchIndexAckEnabled() const;
+
     friend class PulsarWrapper;
+    friend class PulsarFriend;
 
    private:
     std::shared_ptr<ConsumerConfigurationImpl> impl_;