From 475a0adeeca611627891667f45dffdec93805ddc Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Fri, 4 Oct 2019 13:53:24 +0900 Subject: [PATCH] Fix bad_weak_ptr error when closing producer --- pulsar-client-cpp/lib/ConsumerImpl.cc | 10 +++++++--- pulsar-client-cpp/lib/ConsumerImpl.h | 2 +- pulsar-client-cpp/lib/ProducerImpl.cc | 25 +++++++++++++++++-------- pulsar-client-cpp/lib/ProducerImpl.h | 2 +- 4 files changed, 26 insertions(+), 13 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index d2688007f1419..1f27aca119db1 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/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 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; diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index b917f79099dcc..7efc556a561d6 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/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; diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 8488c3d01b161..5ec30204fc47b 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/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(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 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; diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index 3a3780417d30f..7b34519758ebe 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/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);