[feat] Reader support readeNextAsync interface.

include/pulsar/Reader.h

@@ -29,6 +29,7 @@ class PulsarFriend;
 class ReaderImpl;
 
 typedef std::function<void(Result result, bool hasMessageAvailable)> HasMessageAvailableCallback;
+typedef std::function<void(Result result, const Message& message)> ReadNextCallback;
 
 /**
  * A Reader can be used to scan through all the messages currently available in a topic.
@@ -68,6 +69,13 @@ class PULSAR_PUBLIC Reader {
      */
     Result readNext(Message& msg, int timeoutMs);
 
+    /**
+     * Read asynchronously the next message in the topic.
+     *
+     * @param callback
+     */
+    void readNextAsync(ReadNextCallback callback);
+
     /**
      * Close the reader and stop the broker to push more messages
      *

include/pulsar/ReaderConfiguration.h

@@ -162,6 +162,7 @@ class PULSAR_PUBLIC ReaderConfiguration {
      * Set the internal subscription name.
      *
      * @param internal subscriptionName
+     * Default value is reader-{random string}.
      */
     void setInternalSubscriptionName(std::string internalSubscriptionName);
 

lib/ConsumerImpl.cc

@@ -841,7 +841,7 @@ Result ConsumerImpl::receive(Message& msg) {
     return res;
 }
 
-void ConsumerImpl::receiveAsync(ReceiveCallback& callback) {
+void ConsumerImpl::receiveAsync(ReceiveCallback callback) {
     Message msg;
 
     // fail the callback if consumer is closing or closed

lib/ConsumerImpl.h

@@ -94,7 +94,7 @@ class ConsumerImpl : public ConsumerImplBase {
     const std::string& getTopic() const override;
     Result receive(Message& msg) override;
     Result receive(Message& msg, int timeout) override;
-    void receiveAsync(ReceiveCallback& callback) override;
+    void receiveAsync(ReceiveCallback callback) override;
     void unsubscribeAsync(ResultCallback callback) override;
     void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) override;
     void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) override;

lib/ConsumerImplBase.h

@@ -51,7 +51,7 @@ class ConsumerImplBase : public HandlerBase, public std::enable_shared_from_this
     virtual const std::string& getSubscriptionName() const = 0;
     virtual Result receive(Message& msg) = 0;
     virtual Result receive(Message& msg, int timeout) = 0;
-    virtual void receiveAsync(ReceiveCallback& callback) = 0;
+    virtual void receiveAsync(ReceiveCallback callback) = 0;
     void batchReceiveAsync(BatchReceiveCallback callback);
     virtual void unsubscribeAsync(ResultCallback callback) = 0;
     virtual void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) = 0;

lib/MultiTopicsConsumerImpl.cc

@@ -583,7 +583,7 @@ Result MultiTopicsConsumerImpl::receive(Message& msg, int timeout) {
     }
 }
 
-void MultiTopicsConsumerImpl::receiveAsync(ReceiveCallback& callback) {
+void MultiTopicsConsumerImpl::receiveAsync(ReceiveCallback callback) {
     Message msg;
 
     // fail the callback if consumer is closing or closed

lib/MultiTopicsConsumerImpl.h

@@ -65,7 +65,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
     const std::string& getTopic() const override;
     Result receive(Message& msg) override;
     Result receive(Message& msg, int timeout) override;
-    void receiveAsync(ReceiveCallback& callback) override;
+    void receiveAsync(ReceiveCallback callback) override;
     void unsubscribeAsync(ResultCallback callback) override;
     void acknowledgeAsync(const MessageId& msgId, ResultCallback callback) override;
     void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback) override;

lib/Reader.cc

@@ -49,6 +49,14 @@ Result Reader::readNext(Message& msg, int timeoutMs) {
     return impl_->readNext(msg, timeoutMs);
 }
 
+void Reader::readNextAsync(ReadNextCallback callback) {
+    if (!impl_) {
+        return callback(ResultConsumerNotInitialized, {});
+    }
+
+    impl_->readNextAsync(callback);
+}
+
 Result Reader::close() {
     Promise<bool, Result> promise;
     closeAsync(WaitForCallback(promise));

lib/ReaderImpl.cc

@@ -111,6 +111,14 @@ Result ReaderImpl::readNext(Message& msg, int timeoutMs) {
     return res;
 }
 
+void ReaderImpl::readNextAsync(ReceiveCallback callback) {
+    auto self = shared_from_this();
+    consumer_->receiveAsync([self, callback](Result result, const Message& message) {
+        self->acknowledgeIfNecessary(result, message);
+        callback(result, message);
+    });
+}
+
 void ReaderImpl::messageListener(Consumer consumer, const Message& msg) {
     readerListener_(Reader(shared_from_this()), msg);
     acknowledgeIfNecessary(ResultOk, msg);

lib/ReaderImpl.h

@@ -67,6 +67,7 @@ class PULSAR_PUBLIC ReaderImpl : public std::enable_shared_from_this<ReaderImpl>
 
     Result readNext(Message& msg);
     Result readNext(Message& msg, int timeoutMs);
+    void readNextAsync(ReceiveCallback callback);
 
     void closeAsync(ResultCallback callback);
 

tests/ReaderTest.cc

@@ -25,6 +25,7 @@
 
 #include "HttpHelper.h"
 #include "PulsarFriend.h"
+#include "WaitUtils.h"
 #include "lib/ClientConnection.h"
 #include "lib/Latch.h"
 #include "lib/LogUtils.h"
@@ -68,6 +69,50 @@ TEST(ReaderTest, testSimpleReader) {
     client.close();
 }
 
+TEST(ReaderTest, testAsyncRead) {
+    Client client(serviceUrl);
+
+    std::string topicName = "persistent://public/default/test-simple-reader" + std::to_string(time(nullptr));
+
+    ReaderConfiguration readerConf;
+    Reader reader;
+    ASSERT_EQ(ResultOk, client.createReader(topicName, MessageId::earliest(), readerConf, reader));
+
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));
+
+    for (int i = 0; i < 10; i++) {
+        std::string content = "my-message-" + std::to_string(i);
+        Message msg = MessageBuilder().setContent(content).build();
+        ASSERT_EQ(ResultOk, producer.send(msg));
+    }
+
+    for (int i = 0; i < 10; i++) {
+        reader.readNextAsync([i](Result result, const Message& msg) {
+            ASSERT_EQ(ResultOk, result);
+            std::string content = msg.getDataAsString();
+            std::string expected = "my-message-" + std::to_string(i);
+            ASSERT_EQ(expected, content);
+        });
+    }
+
+    waitUntil(
+        std::chrono::seconds(5),
+        [&]() {
+            bool hasMsg;
+            reader.hasMessageAvailable(hasMsg);
+            return !hasMsg;
+        },
+        1000);
+    bool hasMsg;
+    reader.hasMessageAvailable(hasMsg);
+    ASSERT_FALSE(hasMsg);
+
+    producer.close();
+    reader.close();
+    client.close();
+}
+
 TEST(ReaderTest, testReaderAfterMessagesWerePublished) {
     Client client(serviceUrl);