Dlq

CMakeLists.txt

@@ -20,13 +20,13 @@
 cmake_minimum_required(VERSION 3.4)
 
 project (pulsar-cpp)
-set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake_modules")
+set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${PROJECT_SOURCE_DIR}/cmake_modules")
 
-execute_process(COMMAND cat ${CMAKE_SOURCE_DIR}/version.txt OUTPUT_STRIP_TRAILING_WHITESPACE
+execute_process(COMMAND cat ${PROJECT_SOURCE_DIR}/version.txt OUTPUT_STRIP_TRAILING_WHITESPACE
             OUTPUT_VARIABLE PULSAR_CLIENT_VERSION)
 message(STATUS "Pulsar Client version: ${PULSAR_CLIENT_VERSION}")
 
-execute_process(COMMAND ${CMAKE_SOURCE_DIR}/build-support/gen-pulsar-version-macro.py OUTPUT_STRIP_TRAILING_WHITESPACE
+execute_process(COMMAND ${PROJECT_SOURCE_DIR}/build-support/gen-pulsar-version-macro.py OUTPUT_STRIP_TRAILING_WHITESPACE
         OUTPUT_VARIABLE PULSAR_CLIENT_VERSION_MACRO)
 message(STATUS "Pulsar Client version macro: ${PULSAR_CLIENT_VERSION_MACRO}")
 
@@ -36,11 +36,11 @@ configure_file(templates/Version.h.in include/pulsar/Version.h @ONLY)
 option(LINK_STATIC "Link against static libraries" OFF)
 if (VCPKG_TRIPLET)
     message(STATUS "Use vcpkg, triplet is ${VCPKG_TRIPLET}")
-    set(CMAKE_PREFIX_PATH "${CMAKE_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}")
+    set(CMAKE_PREFIX_PATH "${PROJECT_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}")
     message(STATUS "Use CMAKE_PREFIX_PATH: ${CMAKE_PREFIX_PATH}")
     set(PROTOC_PATH "${CMAKE_PREFIX_PATH}/tools/protobuf/protoc")
     message(STATUS "Use protoc: ${PROTOC_PATH}")
-    set(VCPKG_DEBUG_ROOT "${CMAKE_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}/debug")
+    set(VCPKG_DEBUG_ROOT "${PROJECT_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}/debug")
     if (CMAKE_BUILD_TYPE STREQUAL "Debug")
         set(ZLIB_ROOT ${VCPKG_DEBUG_ROOT})
         set(OPENSSL_ROOT_DIR ${VCPKG_DEBUG_ROOT})
@@ -311,13 +311,13 @@ MESSAGE(STATUS "HAS_SNAPPY: ${HAS_SNAPPY}")
 set(ADDITIONAL_LIBRARIES $ENV{PULSAR_ADDITIONAL_LIBRARIES})
 link_directories( $ENV{PULSAR_ADDITIONAL_LIBRARY_PATH} )
 
-set(AUTOGEN_DIR ${CMAKE_BINARY_DIR}/generated)
+set(AUTOGEN_DIR ${PROJECT_BINARY_DIR}/generated)
 file(MAKE_DIRECTORY ${AUTOGEN_DIR})
 
 include_directories(
-  ${CMAKE_SOURCE_DIR}
-  ${CMAKE_SOURCE_DIR}/include
-  ${CMAKE_BINARY_DIR}/include
+  ${PROJECT_SOURCE_DIR}
+  ${PROJECT_SOURCE_DIR}/include
+  ${PROJECT_BINARY_DIR}/include
   ${AUTOGEN_DIR}
   ${Boost_INCLUDE_DIRS}
   ${OPENSSL_INCLUDE_DIR}
@@ -391,7 +391,7 @@ if(NOT APPLE AND NOT MSVC)
     set(COMMON_LIBS ${COMMON_LIBS} rt)
 endif ()
 
-link_directories(${CMAKE_BINARY_DIR}/lib)
+link_directories(${PROJECT_BINARY_DIR}/lib)
 
 set(LIB_NAME $ENV{PULSAR_LIBRARY_NAME})
 if (NOT LIB_NAME)
@@ -421,26 +421,26 @@ if (BUILD_WIRESHARK)
 endif()
 
 find_package(ClangTools)
-set(BUILD_SUPPORT_DIR "${CMAKE_SOURCE_DIR}/build-support")
+set(BUILD_SUPPORT_DIR "${PROJECT_SOURCE_DIR}/build-support")
 add_custom_target(format ${BUILD_SUPPORT_DIR}/run_clang_format.py
         ${CLANG_FORMAT_BIN}
         0
         ${BUILD_SUPPORT_DIR}/clang_format_exclusions.txt
-        ${CMAKE_SOURCE_DIR}/lib
-        ${CMAKE_SOURCE_DIR}/perf
-        ${CMAKE_SOURCE_DIR}/examples
-        ${CMAKE_SOURCE_DIR}/tests
-        ${CMAKE_SOURCE_DIR}/include
-        ${CMAKE_SOURCE_DIR}/wireshark)
+        ${PROJECT_SOURCE_DIR}/lib
+        ${PROJECT_SOURCE_DIR}/perf
+        ${PROJECT_SOURCE_DIR}/examples
+        ${PROJECT_SOURCE_DIR}/tests
+        ${PROJECT_SOURCE_DIR}/include
+        ${PROJECT_SOURCE_DIR}/wireshark)
 
 # `make check-format` option (for CI test)
 add_custom_target(check-format ${BUILD_SUPPORT_DIR}/run_clang_format.py
         ${CLANG_FORMAT_BIN}
         1
         ${BUILD_SUPPORT_DIR}/clang_format_exclusions.txt
-        ${CMAKE_SOURCE_DIR}/lib
-        ${CMAKE_SOURCE_DIR}/perf
-        ${CMAKE_SOURCE_DIR}/examples
-        ${CMAKE_SOURCE_DIR}/tests
-        ${CMAKE_SOURCE_DIR}/include
-        ${CMAKE_SOURCE_DIR}/wireshark)
+        ${PROJECT_SOURCE_DIR}/lib
+        ${PROJECT_SOURCE_DIR}/perf
+        ${PROJECT_SOURCE_DIR}/examples
+        ${PROJECT_SOURCE_DIR}/tests
+        ${PROJECT_SOURCE_DIR}/include
+        ${PROJECT_SOURCE_DIR}/wireshark)

Doxyfile

@@ -1204,6 +1204,23 @@ HTML_EXTRA_STYLESHEET  =
 
 HTML_EXTRA_FILES       =
 
+# The HTML_COLORSTYLE tag can be used to specify if the generated HTML output
+# should be rendered with a dark or light theme. Default setting AUTO_LIGHT
+# enables light output unless the user preference is dark output. Other options
+# are DARK to always use dark mode, LIGHT to always use light mode, AUTO_DARK to
+# default to dark mode unless the user prefers light mode, and TOGGLE to let the
+# user toggle between dark and light mode via a button.
+# Possible values are: LIGHT Always generate light output., DARK Always generate
+# dark output., AUTO_LIGHT Automatically set the mode according to the user
+# preference, use light mode if no preference is set (the default)., AUTO_DARK
+# Automatically set the mode according to the user preference, use dark mode if
+# no preference is set. and TOGGLE Allow to user to switch between light and
+# dark mode via a button..
+# The default value is: AUTO_LIGHT.
+# This tag requires that the tag GENERATE_HTML is set to YES.
+
+HTML_COLORSTYLE        = TOGGLE
+
 # The HTML_COLORSTYLE_HUE tag controls the color of the HTML output. Doxygen
 # will adjust the colors in the style sheet and background images according to
 # this color. Hue is specified as an angle on a colorwheel, see

build-support/download-release-artifacts.py

@@ -45,6 +45,9 @@
     data = json.loads(response.read().decode("utf-8"))
     for artifact in data['artifacts']:
         name = artifact['name']
+        # Skip debug artifact
+        if name.endswith("-Debug"):
+            continue
         url = artifact['archive_download_url']
 
         print('Downloading %s from %s' % (name, url))

build-support/stage-release.sh

@@ -39,6 +39,12 @@ cd $PULSAR_CPP_PATH
 build-support/generate-source-archive.sh $DEST_PATH
 build-support/download-release-artifacts.py $WORKFLOW_ID $DEST_PATH
 
+pushd "$DEST_PATH"
+tar cvzf x64-windows-static.tar.gz x64-windows-static
+tar cvzf x86-windows-static.tar.gz x86-windows-static
+rm -r x64-windows-static x86-windows-static
+popd
+
 # Sign all files
 cd $DEST_PATH
 find . -type f | xargs $PULSAR_CPP_PATH/build-support/sign-files.sh

include/pulsar/ConsumerConfiguration.h

@@ -34,6 +34,7 @@
 #include <memory>
 
 #include "BatchReceivePolicy.h"
+#include "DeadLetterPolicy.h"
 
 namespace pulsar {
 
@@ -398,6 +399,20 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     const BatchReceivePolicy& getBatchReceivePolicy() const;
 
+    /**
+     * Set dead letter policy.
+     *
+     * @param deadLetterPolicy thd default 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`.

include/pulsar/DeadLetterPolicy.h

@@ -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.
+ */
+#ifndef DEAD_LETTER_POLICY_HPP_
+#define DEAD_LETTER_POLICY_HPP_
+
+#include <pulsar/defines.h>
+
+#include <memory>
+
+namespace pulsar {
+
+struct DeadLetterPolicyImpl;
+
+/**
+ * Configuration for the "dead letter queue" feature in consumer.
+ *
+ * see @DeadLetterPolicyBuilder
+ */
+class PULSAR_PUBLIC DeadLetterPolicy {
+   public:
+
+    DeadLetterPolicy();
+
+    /**
+     * Get dead letter topic
+     *
+     * @return
+     */
+    std::string getDeadLetterTopic() const;
+
+    /**
+     * Get max redeliver count
+     *
+     * @return
+     */
+    int getMaxRedeliverCount() const;
+
+    /**
+     * Get initial subscription name
+     *
+     * @return
+     */
+    std::string getInitialSubscriptionName() const;
+
+   private:
+    friend class DeadLetterPolicyBuilder;
+
+    typedef std::shared_ptr<DeadLetterPolicyImpl> DeadLetterPolicyImplPtr;
+    DeadLetterPolicyImplPtr impl_;
+
+    explicit DeadLetterPolicy(const DeadLetterPolicyImplPtr& impl);
+
+};
+}  // namespace pulsar
+
+#endif /* DEAD_LETTER_POLICY_HPP_ */

include/pulsar/DeadLetterPolicyBuilder.h

@@ -0,0 +1,82 @@
+/**
+ * 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 DEAD_LETTER_POLICY_BUILD_HPP_
+#define DEAD_LETTER_POLICY_BUILD_HPP_
+
+#include <pulsar/defines.h>
+#include <pulsar/DeadLetterPolicy.h>
+
+#include <memory>
+
+namespace pulsar {
+
+struct DeadLetterPolicyImpl;
+
+/**
+ * The builder to build a DeadLetterPolicyBuilder
+ *
+ * Example of building DeadLetterPolicy:
+ *
+ * ```c++
+ * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder()
+ *                       .deadLetterTopic("dlq-topic")
+ *                       .maxRedeliverCount(10)
+ *                       .initialSubscriptionName("init-sub-name")
+ *                       .build();
+ * ```
+ */
+class PULSAR_PUBLIC DeadLetterPolicyBuilder {
+   public:
+
+    DeadLetterPolicyBuilder();
+
+    /**
+     * Set dead letter topic
+     *
+     * @return
+     */
+    DeadLetterPolicyBuilder& deadLetterTopic(const std::string& deadLetterTopic);
+
+    /**
+     * Set max redeliver count
+     *
+     * @return
+     */
+    DeadLetterPolicyBuilder& maxRedeliverCount(int maxRedeliverCount);
+
+    /**
+     * Set initial subscription name
+     *
+     * @return
+     */
+    DeadLetterPolicyBuilder& initialSubscriptionName(const std::string& initialSubscriptionName);
+
+    /**
+     * Build DeadLetterPolicy.
+     *
+     * @return
+     */
+    DeadLetterPolicy build();
+
+   private:
+    std::shared_ptr<DeadLetterPolicyImpl> impl_;
+};
+}  // namespace pulsar
+
+#endif /* DEAD_LETTER_POLICY_BUILD_HPP_ */

include/pulsar/Message.h

@@ -90,8 +90,15 @@ class PULSAR_PUBLIC Message {
      * Get string representation of the message
      *
      * @return the string representation of the message payload
+     *
+     * NOTE: For MSVC with debug mode, return a thread local std::string object to avoid memory allocation
+     * across DLLs and applications, which could lead to a crash.
      */
+#if defined(_MSC_VER) && !defined(NDEBUG)
+    const std::string& getDataAsString() const;
+#else
     std::string getDataAsString() const;
+#endif
 
     /**
      * Get key value message.

include/pulsar/MessageId.h

@@ -37,8 +37,12 @@ class PULSAR_PUBLIC MessageId {
     MessageId();
 
     /**
+     * @deprecated
+     *
      * Construct the MessageId
      *
+     * NOTE: This API still exists for backward compatibility, use MessageIdBuilder instead.
+     *
      * @param partition the partition number of a topic
      * @param ledgerId the ledger id
      * @param entryId the entry id
@@ -88,6 +92,7 @@ class PULSAR_PUBLIC MessageId {
     int64_t entryId() const;
     int32_t batchIndex() const;
     int32_t partition() const;
+    int32_t batchSize() const;
 
    private:
     friend class ConsumerImpl;
@@ -102,11 +107,14 @@ class PULSAR_PUBLIC MessageId {
     friend class PulsarWrapper;
     friend class PulsarFriend;
     friend class NegativeAcksTracker;
+    friend class MessageIdBuilder;
 
     friend PULSAR_PUBLIC std::ostream& operator<<(std::ostream& s, const MessageId& messageId);
 
     typedef std::shared_ptr<MessageIdImpl> MessageIdImplPtr;
     MessageIdImplPtr impl_;
+
+    explicit MessageId(const MessageIdImplPtr& impl);
 };
 
 typedef std::vector<MessageId> MessageIdList;