[pulsar-client-cpp] Fix bad_weak_ptr error when closing producer

pulsar-client-cpp/lib/ConsumerImpl.cc

@@ -96,7 +96,6 @@ ConsumerImpl::~ConsumerImpl() {
     incomingMessages_.clear();
     if (state_ == Ready) {
         LOG_WARN(getName() << "Destroyed consumer which was not properly closed");
-        closeAsync(ResultCallback());
     }
 }
 
@@ -818,6 +817,10 @@ void ConsumerImpl::disconnectConsumer() {
 
 void ConsumerImpl::closeAsync(ResultCallback callback) {
     Lock lock(mutex_);
+
+    // Keep a reference to ensure object is kept alive
+    ConsumerImplPtr ptr = shared_from_this();
+
     if (state_ != Ready) {
         lock.unlock();
         if (callback) {
@@ -853,15 +856,16 @@ void ConsumerImpl::closeAsync(ResultCallback callback) {
     Future<Result, ResponseData> future =
         cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
     if (callback) {
+        // Pass the shared pointer "ptr" to the handler to prevent the object from being destroyed
         future.addListener(
-            std::bind(&ConsumerImpl::handleClose, shared_from_this(), std::placeholders::_1, callback));
+            std::bind(&ConsumerImpl::handleClose, shared_from_this(), std::placeholders::_1, callback, ptr));
     }
 
     // fail pendingReceive callback
     failPendingReceiveCallback();
 }
 
-void ConsumerImpl::handleClose(Result result, ResultCallback callback) {
+void ConsumerImpl::handleClose(Result result, ResultCallback callback, ConsumerImplPtr consumer) {
     if (result == ResultOk) {
         Lock lock(mutex_);
         state_ = Closed;

pulsar-client-cpp/lib/ConsumerImpl.h

@@ -120,7 +120,7 @@ class ConsumerImpl : public ConsumerImplBase,
     void handleCreateConsumer(const ClientConnectionPtr& cnx, Result result);
 
     void internalListener();
-    void handleClose(Result result, ResultCallback callback);
+    void handleClose(Result result, ResultCallback callback, ConsumerImplPtr consumer);
     virtual HandlerBaseWeakPtr get_weak_from_this() { return shared_from_this(); }
     virtual const std::string& getName() const;
     virtual int getNumOfPrefetchedMessages() const;

pulsar-client-cpp/lib/ProducerImpl.cc

@@ -78,18 +78,16 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const std::string& topic, const
         std::string logCtx = logCtxStream.str();
         msgCrypto_ = std::make_shared<MessageCrypto>(logCtx, true);
         msgCrypto_->addPublicKeyCipher(conf_.getEncryptionKeys(), conf_.getCryptoKeyReader());
-
-        dataKeyGenTImer_ = executor_->createDeadlineTimer();
-        dataKeyGenTImer_->expires_from_now(boost::posix_time::seconds(dataKeyGenIntervalSec_));
-        dataKeyGenTImer_->async_wait(
-            std::bind(&pulsar::ProducerImpl::refreshEncryptionKey, this, std::placeholders::_1));
     }
 }
 
 ProducerImpl::~ProducerImpl() {
     LOG_DEBUG(getName() << "~ProducerImpl");
-    closeAsync(ResultCallback());
+    cancelTimers();
     printStats();
+    if (state_ == Ready) {
+        LOG_WARN(getName() << "Destroyed producer which was not properly closed");
+    }
 }
 
 const std::string& ProducerImpl::getTopic() const { return topic_; }
@@ -171,6 +169,13 @@ void ProducerImpl::handleCreateProducer(const ClientConnectionPtr& cnx, Result r
         backoff_.reset();
         lock.unlock();
 
+        if (!dataKeyGenTImer_ && conf_.isEncryptionEnabled()) {
+            dataKeyGenTImer_ = executor_->createDeadlineTimer();
+            dataKeyGenTImer_->expires_from_now(boost::posix_time::seconds(dataKeyGenIntervalSec_));
+            dataKeyGenTImer_->async_wait(std::bind(&pulsar::ProducerImpl::refreshEncryptionKey,
+                                                   shared_from_this(), std::placeholders::_1));
+        }
+
         // Initialize the sendTimer only once per producer and only when producer timeout is
         // configured. Set the timeout as configured value and asynchronously wait for the
         // timeout to happen.
@@ -470,6 +475,9 @@ void ProducerImpl::printStats() {
 void ProducerImpl::closeAsync(CloseCallback callback) {
     Lock lock(mutex_);
 
+    // Keep a reference to ensure object is kept alive
+    ProducerImplPtr ptr = shared_from_this();
+
     cancelTimers();
 
     if (state_ != Ready) {
@@ -508,12 +516,13 @@ void ProducerImpl::closeAsync(CloseCallback callback) {
     Future<Result, ResponseData> future =
         cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId);
     if (callback) {
+        // Pass the shared pointer "ptr" to the handler to prevent the object from being destroyed
         future.addListener(
-            std::bind(&ProducerImpl::handleClose, shared_from_this(), std::placeholders::_1, callback));
+            std::bind(&ProducerImpl::handleClose, shared_from_this(), std::placeholders::_1, callback, ptr));
     }
 }
 
-void ProducerImpl::handleClose(Result result, ResultCallback callback) {
+void ProducerImpl::handleClose(Result result, ResultCallback callback, ProducerImplPtr producer) {
     if (result == ResultOk) {
         Lock lock(mutex_);
         state_ = Closed;

pulsar-client-cpp/lib/ProducerImpl.h

@@ -131,7 +131,7 @@ class ProducerImpl : public HandlerBase,
 
     void statsCallBackHandler(Result, const Message&, SendCallback, boost::posix_time::ptime);
 
-    void handleClose(Result result, ResultCallback callback);
+    void handleClose(Result result, ResultCallback callback, ProducerImplPtr producer);
 
     void resendMessages(ClientConnectionPtr cnx);