Restore the shutdown virtual methods for cleanup in ClientImpl

Author: BewareMyPower

lib/ClientImpl.cc

@@ -739,15 +739,25 @@ void ClientImpl::shutdown() {
     auto producers = producers_.move();
     auto consumers = consumers_.move();
 
-    auto numProducers = producers.size();
-    auto numConsumers = consumers.size();
-    // Call the destructors out of the lock to avoid deadlocks
-    producers.clear();
-    consumers.clear();
+    for (auto&& kv : producers) {
+        ProducerImplBasePtr producer = kv.second.lock();
+        if (producer) {
+            producer->shutdown();
+        }
+    }
 
-    if (numProducers + numConsumers > 0) {
-        LOG_DEBUG(numProducers << " producers and " << numConsumers << " consumers have been shutdown.");
+    for (auto&& kv : consumers) {
+        ConsumerImplBasePtr consumer = kv.second.lock();
+        if (consumer) {
+            consumer->shutdown();
+        }
     }
+
+    if (producers.size() + consumers.size() > 0) {
+        LOG_DEBUG(producers.size() << " producers and " << consumers.size()
+                                   << " consumers have been shutdown.");
+    }
+
     if (!pool_.close()) {
         // pool_ has already been closed. It means shutdown() has been called before.
         return;

lib/ConsumerImpl.cc

@@ -179,7 +179,7 @@ ConsumerImpl::~ConsumerImpl() {
             LOG_WARN(consumerStr_ << "Client is destroyed and cannot send the CloseConsumer command");
         }
     }
-    shutdown();
+    internalShutdown();
 }
 
 void ConsumerImpl::setPartitionIndex(int partitionIndex) { partitionIndex_ = partitionIndex; }
@@ -373,7 +373,7 @@ void ConsumerImpl::unsubscribeAsync(const ResultCallback& originalCallback) {
 
     auto callback = [this, originalCallback](Result result) {
         if (result == ResultOk) {
-            shutdown();
+            internalShutdown();
             LOG_INFO(getName() << "Unsubscribed successfully");
         } else {
             state_ = Ready;
@@ -1312,7 +1312,7 @@ void ConsumerImpl::disconnectConsumer(const boost::optional<std::string>& assign
 
 void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
     auto callback = [this, originalCallback](Result result, bool alreadyClosed = false) {
-        shutdown();
+        internalShutdown();
         if (result == ResultOk) {
             if (!alreadyClosed) {
                 LOG_INFO(getName() << "Closed consumer " << consumerId_);
@@ -1368,7 +1368,9 @@ void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
 
 const std::string& ConsumerImpl::getName() const { return consumerStr_; }
 
-void ConsumerImpl::shutdown() {
+void ConsumerImpl::shutdown() { internalShutdown(); }
+
+void ConsumerImpl::internalShutdown() {
     ackGroupingTrackerPtr_.reset();
     incomingMessages_.clear();
     possibleSendToDeadLetterTopicMessages_.clear();

lib/ConsumerImpl.h

@@ -121,7 +121,8 @@ class ConsumerImpl : public ConsumerImplBase {
     void acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) override;
     void closeAsync(const ResultCallback& callback) override;
     void start() override;
-    void shutdown();
+    void shutdown() override;
+    void internalShutdown();
     bool isClosed() override;
     bool isOpen() override;
     Result pauseMessageListener() override;

lib/ConsumerImplBase.h

@@ -63,6 +63,7 @@ class ConsumerImplBase : public HandlerBase {
     virtual void acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) = 0;
     virtual void closeAsync(const ResultCallback& callback) = 0;
     virtual void start() = 0;
+    virtual void shutdown() = 0;
     virtual bool isClosed() = 0;
     virtual bool isOpen() = 0;
     virtual Result pauseMessageListener() = 0;

lib/MultiTopicsConsumerImpl.cc

@@ -328,7 +328,7 @@ void MultiTopicsConsumerImpl::unsubscribeAsync(const ResultCallback& originalCal
 
     auto callback = [this, originalCallback](Result result) {
         if (result == ResultOk) {
-            shutdown();
+            internalShutdown();
             LOG_INFO(getName() << "Unsubscribed successfully");
         } else {
             state_ = Ready;
@@ -452,7 +452,7 @@ void MultiTopicsConsumerImpl::closeAsync(const ResultCallback& originalCallback)
     auto callback = [weakSelf, originalCallback](Result result) {
         auto self = weakSelf.lock();
         if (self) {
-            self->shutdown();
+            self->internalShutdown();
             if (result != ResultOk) {
                 LOG_WARN(self->getName() << "Failed to close consumer: " << result);
                 if (result != ResultAlreadyClosed) {
@@ -735,7 +735,7 @@ void MultiTopicsConsumerImpl::negativeAcknowledge(const MessageId& msgId) {
     }
 }
 
-MultiTopicsConsumerImpl::~MultiTopicsConsumerImpl() { shutdown(); }
+MultiTopicsConsumerImpl::~MultiTopicsConsumerImpl() { internalShutdown(); }
 
 Future<Result, ConsumerImplBaseWeakPtr> MultiTopicsConsumerImpl::getConsumerCreatedFuture() {
     return multiTopicsConsumerCreatedPromise_.getFuture();
@@ -746,7 +746,9 @@ const std::string& MultiTopicsConsumerImpl::getTopic() const { return topic(); }
 
 const std::string& MultiTopicsConsumerImpl::getName() const { return consumerStr_; }
 
-void MultiTopicsConsumerImpl::shutdown() {
+void MultiTopicsConsumerImpl::shutdown() { internalShutdown(); }
+
+void MultiTopicsConsumerImpl::internalShutdown() {
     cancelTimers();
     incomingMessages_.clear();
     topicsPartitions_.clear();

lib/MultiTopicsConsumerImpl.h

@@ -81,7 +81,8 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase {
     void acknowledgeCumulativeAsync(const MessageId& msgId, const ResultCallback& callback) override;
     void closeAsync(const ResultCallback& callback) override;
     void start() override;
-    void shutdown();
+    void shutdown() override;
+    void internalShutdown();
     bool isClosed() override;
     bool isOpen() override;
     Result pauseMessageListener() override;

lib/PartitionedProducerImpl.cc

@@ -79,7 +79,7 @@ MessageRoutingPolicyPtr PartitionedProducerImpl::getMessageRouter() {
     }
 }
 
-PartitionedProducerImpl::~PartitionedProducerImpl() { shutdown(); }
+PartitionedProducerImpl::~PartitionedProducerImpl() { internalShutdown(); }
 // override
 const std::string& PartitionedProducerImpl::getTopic() const { return topic_; }
 
@@ -244,7 +244,9 @@ void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callbac
 }
 
 // override
-void PartitionedProducerImpl::shutdown() {
+void PartitionedProducerImpl::shutdown() { internalShutdown(); }
+
+void PartitionedProducerImpl::internalShutdown() {
     cancelTimers();
     interceptors_->close();
     auto client = client_.lock();
@@ -284,7 +286,7 @@ int64_t PartitionedProducerImpl::getLastSequenceId() const {
 void PartitionedProducerImpl::closeAsync(CloseCallback originalCallback) {
     auto closeCallback = [this, originalCallback](Result result) {
         if (result == ResultOk) {
-            shutdown();
+            internalShutdown();
         }
         if (originalCallback) {
             originalCallback(result);

lib/PartitionedProducerImpl.h

@@ -76,7 +76,8 @@ class PartitionedProducerImpl : public ProducerImplBase,
      */
     void closeAsync(CloseCallback callback) override;
     void start() override;
-    void shutdown();
+    void shutdown() override;
+    void internalShutdown();
     bool isClosed() override;
     const std::string& getTopic() const override;
     Future<Result, ProducerImplBaseWeakPtr> getProducerCreatedFuture() override;

lib/PatternMultiTopicsConsumerImpl.cc

@@ -242,9 +242,9 @@ void PatternMultiTopicsConsumerImpl::start() {
     }
 }
 
-void PatternMultiTopicsConsumerImpl::shutdown() {
+PatternMultiTopicsConsumerImpl::~PatternMultiTopicsConsumerImpl() {
     cancelTimers();
-    MultiTopicsConsumerImpl::shutdown();
+    internalShutdown();
 }
 
 void PatternMultiTopicsConsumerImpl::closeAsync(const ResultCallback& callback) {

lib/PatternMultiTopicsConsumerImpl.h

@@ -54,6 +54,7 @@ class PatternMultiTopicsConsumerImpl : public MultiTopicsConsumerImpl {
                                    const std::string& subscriptionName, const ConsumerConfiguration& conf,
                                    const LookupServicePtr& lookupServicePtr_,
                                    const ConsumerInterceptorsPtr& interceptors);
+    ~PatternMultiTopicsConsumerImpl() override;
 
     const PULSAR_REGEX_NAMESPACE::regex getPattern();
 
@@ -67,9 +68,8 @@ class PatternMultiTopicsConsumerImpl : public MultiTopicsConsumerImpl {
     static NamespaceTopicsPtr topicsListsMinus(std::vector<std::string>& list1,
                                                std::vector<std::string>& list2);
 
-    virtual void closeAsync(const ResultCallback& callback);
-    virtual void start();
-    virtual void shutdown();
+    void closeAsync(const ResultCallback& callback) override;
+    void start() override;
 
    private:
     const std::string patternString_;