Auto publish

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

@@ -57,7 +57,7 @@ jobs:
           sudo curl -o /gtest-parallel https://raw.githubusercontent.com/google/gtest-parallel/master/gtest_parallel.py 
 
       - name: CMake
-        run: cmake . -DBUILD_PERF_TOOLS=ON
+        run: cmake . -DCMAKE_BUILD_TYPE=Debug -DBUILD_PERF_TOOLS=ON
 
       - name: Check formatting
         run: make check-format

CMakeLists.txt

@@ -17,7 +17,7 @@
 # under the License.
 #
 
-cmake_minimum_required(VERSION 3.4)
+cmake_minimum_required(VERSION 3.13)
 
 project (pulsar-cpp)
 set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${PROJECT_SOURCE_DIR}/cmake_modules")
@@ -290,7 +290,7 @@ if (NOT APPLE AND NOT MSVC)
     # Hide all non-exported symbols to avoid conflicts
     add_compile_options(-fvisibility=hidden) 
     if (CMAKE_COMPILER_IS_GNUCC)
-        add_compile_options(-Wl,--exclude-libs,ALL)
+        add_link_options(-Wl,--exclude-libs=ALL)
     endif ()
 endif ()
 

README.md

@@ -30,7 +30,7 @@ Pulsar C++ client uses [doxygen](https://www.doxygen.nl) to build API documents.
 ## Requirements
 
 * A C++ compiler that supports C++11, like GCC >= 4.8
-* CMake >= 3.4
+* CMake >= 3.13
 * [Boost](http://www.boost.org/)
 * [Protocol Buffer](https://developers.google.com/protocol-buffers/) >= 3
 * [libcurl](https://curl.se/libcurl/)

build-support/run_clang_format.py

@@ -46,7 +46,7 @@
     for directory, subdirs, files in os.walk(source_dir):
         for name in files:
             name = os.path.join(directory, name)
-            if not (name.endswith('.h') or name.endswith('.cc')):
+            if not (name.endswith('.h') or name.endswith('.cc') or name.endswith('.c')):
                 continue
 
             excluded = False

examples/SampleAsyncConsumerCApi.c

@@ -17,8 +17,8 @@
  * under the License.
  */
 
-#include <stdio.h>
 #include <pulsar/c/client.h>
+#include <stdio.h>
 
 static void receive_callback(pulsar_result result, pulsar_message_t *message, void *ctx) {
     pulsar_consumer_t *consumer = (pulsar_consumer_t *)ctx;

examples/SampleConsumerCApi.c

@@ -17,8 +17,8 @@
  * under the License.
  */
 
-#include <stdio.h>
 #include <pulsar/c/client.h>
+#include <stdio.h>
 
 int main() {
     pulsar_client_configuration_t *conf = pulsar_client_configuration_create();
@@ -28,7 +28,8 @@ int main() {
     pulsar_consumer_configuration_set_consumer_type(consumer_conf, pulsar_ConsumerShared);
 
     pulsar_consumer_t *consumer;
-    pulsar_result res = pulsar_client_subscribe(client, "my-topic", "my-subscrition", consumer_conf, &consumer);
+    pulsar_result res =
+        pulsar_client_subscribe(client, "my-topic", "my-subscrition", consumer_conf, &consumer);
     if (res != pulsar_result_Ok) {
         printf("Failed to create subscribe to topic: %s\n", pulsar_result_str(res));
         return 1;
@@ -43,7 +44,7 @@ int main() {
         }
 
         printf("Received message with payload: '%.*s'\n", pulsar_message_get_length(message),
-               (const char*)pulsar_message_get_data(message));
+               (const char *)pulsar_message_get_data(message));
 
         pulsar_consumer_acknowledge(consumer, message);
         pulsar_message_free(message);

examples/SampleConsumerListenerCApi.c

@@ -17,12 +17,12 @@
  * under the License.
  */
 
-#include <stdio.h>
 #include <pulsar/c/client.h>
+#include <stdio.h>
 
-static void listener_callback(pulsar_consumer_t* consumer, pulsar_message_t* message, void* ctx) {
+static void listener_callback(pulsar_consumer_t *consumer, pulsar_message_t *message, void *ctx) {
     printf("Received message with payload: '%.*s'\n", pulsar_message_get_length(message),
-           (const char*)pulsar_message_get_data(message));
+           (const char *)pulsar_message_get_data(message));
 
     pulsar_consumer_acknowledge(consumer, message);
     pulsar_message_free(message);
@@ -37,7 +37,8 @@ int main() {
     pulsar_consumer_configuration_set_message_listener(consumer_conf, listener_callback, NULL);
 
     pulsar_consumer_t *consumer;
-    pulsar_result res = pulsar_client_subscribe(client, "my-topic", "my-subscrition", consumer_conf, &consumer);
+    pulsar_result res =
+        pulsar_client_subscribe(client, "my-topic", "my-subscrition", consumer_conf, &consumer);
     if (res != pulsar_result_Ok) {
         printf("Failed to create subscribe to topic: %s\n", pulsar_result_str(res));
         return 1;

examples/SampleProducerCApi.c

@@ -18,18 +18,17 @@
  */
 
 #include <pulsar/c/client.h>
-
 #include <stdio.h>
 #include <string.h>
 
 int main() {
-    pulsar_client_configuration_t *conf = pulsar_client_configuration_create();
+    pulsar_client_configuration_t* conf = pulsar_client_configuration_create();
     pulsar_client_configuration_set_memory_limit(conf, 64 * 1024 * 1024);
-    pulsar_client_t *client = pulsar_client_create("pulsar://localhost:6650", conf);
+    pulsar_client_t* client = pulsar_client_create("pulsar://localhost:6650", conf);
 
     pulsar_producer_configuration_t* producer_conf = pulsar_producer_configuration_create();
     pulsar_producer_configuration_set_batching_enabled(producer_conf, 1);
-    pulsar_producer_t *producer;
+    pulsar_producer_t* producer;
 
     pulsar_result err = pulsar_client_create_producer(client, "my-topic", producer_conf, &producer);
     if (err != pulsar_result_Ok) {

examples/SampleReaderCApi.c

@@ -17,8 +17,8 @@
  * under the License.
  */
 
-#include <stdio.h>
 #include <pulsar/c/client.h>
+#include <stdio.h>
 
 int main() {
     pulsar_client_configuration_t *conf = pulsar_client_configuration_create();
@@ -27,8 +27,8 @@ int main() {
     pulsar_reader_configuration_t *reader_conf = pulsar_reader_configuration_create();
 
     pulsar_reader_t *reader;
-    pulsar_result res = pulsar_client_create_reader(client, "my-topic", pulsar_message_id_earliest(), reader_conf,
-                                                    &reader);
+    pulsar_result res =
+        pulsar_client_create_reader(client, "my-topic", pulsar_message_id_earliest(), reader_conf, &reader);
     if (res != pulsar_result_Ok) {
         printf("Failed to create reader: %s\n", pulsar_result_str(res));
         return 1;
@@ -43,7 +43,7 @@ int main() {
         }
 
         printf("Received message with payload: '%.*s'\n", pulsar_message_get_length(message),
-               (const char*)pulsar_message_get_data(message));
+               (const char *)pulsar_message_get_data(message));
 
         pulsar_message_free(message);
     }

include/pulsar/Message.h

@@ -188,8 +188,7 @@ class PULSAR_PUBLIC Message {
     MessageImplPtr impl_;
 
     Message(MessageImplPtr& impl);
-    Message(const proto::CommandMessage& msg, proto::MessageMetadata& data, SharedBuffer& payload,
-            int32_t partition);
+    Message(const MessageId& messageId, proto::MessageMetadata& metadata, SharedBuffer& payload);
     /// Used for Batch Messages
     Message(const MessageId& messageId, proto::MessageMetadata& metadata, SharedBuffer& payload,
             proto::SingleMessageMetadata& singleMetadata, const std::string& topicName);

include/pulsar/MessageId.h

@@ -108,6 +108,7 @@ class PULSAR_PUBLIC MessageId {
     friend class PulsarFriend;
     friend class NegativeAcksTracker;
     friend class MessageIdBuilder;
+    friend class ChunkMessageIdImpl;
 
     friend PULSAR_PUBLIC std::ostream& operator<<(std::ostream& s, const MessageId& messageId);
 

include/pulsar/Result.h

@@ -92,6 +92,9 @@ enum Result
     ResultMemoryBufferIsFull,  /// Client-wide memory limit has been reached
 
     ResultInterrupted,  /// Interrupted while waiting to dequeue
+
+    ResultDisconnected,  /// Client connection has been disconnected
+    ResultNotFound       /// The generic was not found
 };
 
 // Return string representation of result code

include/pulsar/Schema.h

@@ -134,6 +134,8 @@ enum SchemaType
 // Return string representation of result code
 PULSAR_PUBLIC const char *strSchemaType(SchemaType schemaType);
 
+PULSAR_PUBLIC SchemaType enumSchemaType(std::string schemaTypeStr);
+
 class SchemaInfoImpl;
 
 typedef std::map<std::string, std::string> StringMap;
@@ -195,7 +197,6 @@ class PULSAR_PUBLIC SchemaInfo {
    private:
     typedef std::shared_ptr<SchemaInfoImpl> SchemaInfoImplPtr;
     SchemaInfoImplPtr impl_;
-    static constexpr uint32_t INVALID_SIZE = 0xFFFFFFFF;
 };
 
 }  // namespace pulsar

lib/BatchMessageAcker.h

@@ -31,40 +31,55 @@ class BatchMessageAcker;
 using BatchMessageAckerPtr = std::shared_ptr<BatchMessageAcker>;
 
 class BatchMessageAcker {
+   public:
+    virtual ~BatchMessageAcker() {}
+    // Return false for these methods so that batch index ACK will be falled back to if the acker is created
+    // by deserializing from raw bytes.
+    virtual bool ackIndividual(int32_t) { return false; }
+    virtual bool ackCumulative(int32_t) { return false; }
+
+    bool shouldAckPreviousMessageId() noexcept {
+        bool expectedValue = false;
+        return prevBatchCumulativelyAcked_.compare_exchange_strong(expectedValue, true);
+    }
+
+   private:
+    // When a batched message is acknowledged cumulatively, the previous message id will be acknowledged
+    // without batch index ACK enabled. However, it should be acknowledged only once. Use this flag to
+    // determine whether to acknowledge the previous message id.
+    std::atomic_bool prevBatchCumulativelyAcked_{false};
+};
+
+class BatchMessageAckerImpl : public BatchMessageAcker {
    public:
     using Lock = std::lock_guard<std::mutex>;
 
     static BatchMessageAckerPtr create(int32_t batchSize) {
-        return std::make_shared<BatchMessageAcker>(batchSize);
+        if (batchSize > 0) {
+            return std::make_shared<BatchMessageAckerImpl>(batchSize);
+        } else {
+            return std::make_shared<BatchMessageAcker>();
+        }
     }
 
-    BatchMessageAcker(int32_t batchSize) : bitSet_(batchSize) { bitSet_.set(0, batchSize); }
+    BatchMessageAckerImpl(int32_t batchSize) : bitSet_(batchSize) { bitSet_.set(0, batchSize); }
 
-    bool ackIndividual(int32_t batchIndex) {
+    bool ackIndividual(int32_t batchIndex) override {
         Lock lock{mutex_};
         bitSet_.clear(batchIndex);
         return bitSet_.isEmpty();
     }
 
-    bool ackCumulative(int32_t batchIndex) {
+    bool ackCumulative(int32_t batchIndex) override {
         Lock lock{mutex_};
         // The range of cumulative acknowledgment is closed while BitSet::clear accepts a left-closed
         // right-open range.
         bitSet_.clear(0, batchIndex + 1);
         return bitSet_.isEmpty();
     }
 
-    bool shouldAckPreviousMessageId() noexcept {
-        bool expectedValue = false;
-        return prevBatchCumulativelyAcked_.compare_exchange_strong(expectedValue, true);
-    }
-
    private:
     BitSet bitSet_;
-    // When a batched message is acknowledged cumulatively, the previous message id will be acknowledged
-    // without batch index ACK enabled. However, it should be acknowledged only once. Use this flag to
-    // determine whether to acknowledge the previous message id.
-    std::atomic_bool prevBatchCumulativelyAcked_{false};
     mutable std::mutex mutex_;
 };
 

lib/BinaryProtoLookupService.cc

@@ -155,6 +155,43 @@ Future<Result, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespac
     return promise->getFuture();
 }
 
+Future<Result, boost::optional<SchemaInfo>> BinaryProtoLookupService::getSchema(
+    const TopicNamePtr& topicName) {
+    GetSchemaPromisePtr promise = std::make_shared<Promise<Result, boost::optional<SchemaInfo>>>();
+
+    if (!topicName) {
+        promise->setFailed(ResultInvalidTopicName);
+        return promise->getFuture();
+    }
+    cnxPool_.getConnectionAsync(serviceNameResolver_.resolveHost())
+        .addListener(std::bind(&BinaryProtoLookupService::sendGetSchemaRequest, this, topicName->toString(),
+                               std::placeholders::_1, std::placeholders::_2, promise));
+
+    return promise->getFuture();
+}
+
+void BinaryProtoLookupService::sendGetSchemaRequest(const std::string& topicName, Result result,
+                                                    const ClientConnectionWeakPtr& clientCnx,
+                                                    GetSchemaPromisePtr promise) {
+    if (result != ResultOk) {
+        promise->setFailed(result);
+        return;
+    }
+
+    ClientConnectionPtr conn = clientCnx.lock();
+    uint64_t requestId = newRequestId();
+    LOG_DEBUG("sendGetSchemaRequest. requestId: " << requestId << " topicName: " << topicName);
+
+    conn->newGetSchema(topicName, requestId)
+        .addListener([promise](Result result, boost::optional<SchemaInfo> schemaInfo) {
+            if (result != ResultOk) {
+                promise->setFailed(result);
+                return;
+            }
+            promise->setValue(schemaInfo);
+        });
+}
+
 void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string& nsName, Result result,
                                                                const ClientConnectionWeakPtr& clientCnx,
                                                                NamespaceTopicsPromisePtr promise) {

lib/BinaryProtoLookupService.h

@@ -20,6 +20,7 @@
 #define _PULSAR_BINARY_LOOKUP_SERVICE_HEADER_
 
 #include <pulsar/Authentication.h>
+#include <pulsar/Schema.h>
 
 #include <mutex>
 
@@ -32,6 +33,7 @@ class ConnectionPool;
 class LookupDataResult;
 class ServiceNameResolver;
 using NamespaceTopicsPromisePtr = std::shared_ptr<Promise<Result, NamespaceTopicsPtr>>;
+using GetSchemaPromisePtr = std::shared_ptr<Promise<Result, boost::optional<SchemaInfo>>>;
 
 class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
    public:
@@ -45,6 +47,8 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
 
     Future<Result, NamespaceTopicsPtr> getTopicsOfNamespaceAsync(const NamespaceNamePtr& nsName) override;
 
+    Future<Result, boost::optional<SchemaInfo>> getSchema(const TopicNamePtr& topicName) override;
+
    private:
     std::mutex mutex_;
     uint64_t requestIdGenerator_ = 0;
@@ -68,6 +72,9 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
                                          const ClientConnectionWeakPtr& clientCnx,
                                          NamespaceTopicsPromisePtr promise);
 
+    void sendGetSchemaRequest(const std::string& topiName, Result result,
+                              const ClientConnectionWeakPtr& clientCnx, GetSchemaPromisePtr promise);
+
     void getTopicsOfNamespaceListener(Result result, NamespaceTopicsPtr topicsPtr,
                                       NamespaceTopicsPromisePtr promise);
 

lib/CMakeLists.txt

@@ -73,7 +73,7 @@ if (BUILD_DYNAMIC_LIB)
 endif()
 
 include(CheckCXXSymbolExists)
-check_cxx_symbol_exists(getauxval auvx.h HAVE_AUXV_GETAUXVAL)
+check_cxx_symbol_exists(getauxval sys/auxv.h HAVE_AUXV_GETAUXVAL)
 if(HAVE_AUXV_GETAUXVAL)
     add_definitions(-DPULSAR_AUXV_GETAUXVAL_PRESENT)
 endif()