Fix getLastMessageId method dead recursion. (#117)

Author: shibd

lib/Consumer.cc

@@ -295,7 +295,8 @@ void Consumer::getLastMessageIdAsync(GetLastMessageIdCallback callback) {
         callback(ResultConsumerNotInitialized, MessageId());
         return;
     }
-    getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) {
+
+    impl_->getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) {
         callback(result, response.getLastMessageId());
     });
 }

lib/ConsumerImpl.cc

@@ -1343,7 +1343,7 @@ void ConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback
     if (state == Closed || state == Closing) {
         LOG_ERROR(getName() << "Client connection already closed.");
         if (callback) {
-            callback(ResultAlreadyClosed, MessageId());
+            callback(ResultAlreadyClosed, GetLastMessageIdResponse());
         }
         return;
     }

lib/ConsumerImpl.h

@@ -44,7 +44,6 @@ class BatchAcknowledgementTracker;
 class MessageCrypto;
 class GetLastMessageIdResponse;
 typedef std::shared_ptr<MessageCrypto> MessageCryptoPtr;
-typedef std::function<void(Result, const GetLastMessageIdResponse&)> BrokerGetLastMessageIdCallback;
 typedef std::shared_ptr<Backoff> BackoffPtr;
 
 class AckGroupingTracker;
@@ -124,6 +123,7 @@ class ConsumerImpl : public ConsumerImplBase {
     const std::string& getName() const override;
     int getNumOfPrefetchedMessages() const override;
     void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) override;
+    void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) override;
     void seekAsync(const MessageId& msgId, ResultCallback callback) override;
     void seekAsync(uint64_t timestamp, ResultCallback callback) override;
     void negativeAcknowledge(const MessageId& msgId) override;
@@ -139,7 +139,6 @@ class ConsumerImpl : public ConsumerImplBase {
 
     virtual bool isReadCompacted();
     virtual void hasMessageAvailableAsync(HasMessageAvailableCallback callback);
-    virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback);
     void beforeConnectionChange(ClientConnection& cnx) override;
 
    protected:

lib/ConsumerImplBase.h

@@ -25,12 +25,12 @@
 #include <set>
 
 #include "Future.h"
+#include "GetLastMessageIdResponse.h"
 #include "HandlerBase.h"
 
 namespace pulsar {
 class ConsumerImplBase;
 using ConsumerImplBaseWeakPtr = std::weak_ptr<ConsumerImplBase>;
-
 class OpBatchReceive {
    public:
     OpBatchReceive();
@@ -68,6 +68,7 @@ class ConsumerImplBase : public HandlerBase, public std::enable_shared_from_this
     virtual void redeliverUnacknowledgedMessages(const std::set<MessageId>& messageIds) = 0;
     virtual int getNumOfPrefetchedMessages() const = 0;
     virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) = 0;
+    virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) = 0;
     virtual void seekAsync(const MessageId& msgId, ResultCallback callback) = 0;
     virtual void seekAsync(uint64_t timestamp, ResultCallback callback) = 0;
     virtual void negativeAcknowledge(const MessageId& msgId) = 0;

lib/GetLastMessageIdResponse.h

@@ -19,6 +19,7 @@
 #pragma once
 
 #include <pulsar/MessageId.h>
+#include <pulsar/Result.h>
 
 #include <iostream>
 
@@ -54,4 +55,6 @@ class GetLastMessageIdResponse {
     bool hasMarkDeletePosition_;
 };
 
+typedef std::function<void(Result, const GetLastMessageIdResponse&)> BrokerGetLastMessageIdCallback;
+
 }  // namespace pulsar

lib/MultiTopicsConsumerImpl.cc

@@ -803,6 +803,10 @@ void MultiTopicsConsumerImpl::getBrokerConsumerStatsAsync(BrokerConsumerStatsCal
     });
 }
 
+void MultiTopicsConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) {
+    callback(ResultOperationNotSupported, GetLastMessageIdResponse());
+}
+
 void MultiTopicsConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerStats brokerConsumerStats,
                                                      LatchPtr latchPtr,
                                                      MultiTopicsBrokerConsumerStatsPtr statsPtr, size_t index,

lib/MultiTopicsConsumerImpl.h

@@ -82,6 +82,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
     const std::string& getName() const override;
     int getNumOfPrefetchedMessages() const override;
     void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback) override;
+    void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback) override;
     void seekAsync(const MessageId& msgId, ResultCallback callback) override;
     void seekAsync(uint64_t timestamp, ResultCallback callback) override;
     void negativeAcknowledge(const MessageId& msgId) override;

tests/ConsumerTest.cc

@@ -841,6 +841,28 @@ TEST(ConsumerTest, testPartitionsWithCloseUnblock) {
     thread.join();
 }
 
+TEST(ConsumerTest, testGetLastMessageId) {
+    Client client(lookupUrl);
+    const std::string topic = "testGetLastMessageId-" + std::to_string(time(nullptr));
+
+    Consumer consumer;
+    ASSERT_EQ(ResultOk, client.subscribe(topic, "test-sub", consumer));
+
+    MessageId msgId;
+    ASSERT_EQ(ResultOk, consumer.getLastMessageId(msgId));
+    ASSERT_EQ(msgId, MessageId(-1, -1, -1, -1));
+
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
+    Message msg = MessageBuilder().setContent("message").build();
+    ASSERT_EQ(ResultOk, producer.send(msg));
+
+    ASSERT_EQ(ResultOk, consumer.getLastMessageId(msgId));
+    ASSERT_NE(msgId, MessageId(-1, -1, -1, -1));
+
+    client.close();
+}
+
 TEST(ConsumerTest, testGetLastMessageIdBlockWhenConnectionDisconnected) {
     int operationTimeout = 5;
     ClientConfiguration clientConfiguration;