CPP Client - Accessing connection_ pointer under a lock (apache#555)

afalko · Jul 10, 2017 · 365230c · 365230c
1 parent 9a8098f
commit 365230c
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 3 deletions.
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -155,9 +155,9 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r
             firstTime = false;
         }
         LOG_INFO(getName() << "Created consumer on broker " << cnx->cnxString());
-        connection_ = cnx;
         {
             Lock lock(mutex_);
+            connection_ = cnx;
             incomingMessages_.clear();
             cnx->registerConsumer(consumerId_, shared_from_this());
             state_ = Ready;
@@ -581,7 +581,9 @@ void ConsumerImpl::doAcknowledge(const BatchMessageId& messageId, proto::Command
 
 void ConsumerImpl::disconnectConsumer() {
     LOG_DEBUG("Broker notification of Closed consumer: " << consumerId_);
+    Lock lock(mutex_);
     connection_.reset();
+    lock.unlock();
     scheduleReconnection(shared_from_this());
 }
 

diff --git a/pulsar-client-cpp/lib/HandlerBase.cc b/pulsar-client-cpp/lib/HandlerBase.cc
@@ -48,11 +48,13 @@ void HandlerBase::start() {
 }
 
 void HandlerBase::grabCnx() {
+    Lock lock(mutex_);
     if (connection_.lock()) {
+        lock.unlock();
         LOG_INFO(getName() << "Ignoring reconnection request since we're already connected");
         return;
     }
-
+    lock.unlock();
     LOG_INFO(getName() << "Getting connection from pool");
     ClientImplPtr client = client_.lock();
     Future<Result, ClientConnectionWeakPtr> future = client->getConnection(topic_);

diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc
@@ -548,9 +548,10 @@ bool ProducerImpl::ackReceived(uint64_t sequenceId) {
 
 void ProducerImpl::disconnectProducer() {
 	LOG_DEBUG("Broker notification of Closed producer: " << producerId_);
+	Lock lock(mutex_);
 	connection_.reset();
+	lock.unlock();
 	scheduleReconnection(shared_from_this());
-
 }
 
 const std::string& ProducerImpl::getName() const{