CPP Client - Reducing number of duplicate lookups (apache#556)

* CPP Client: Avoiding duplicate reconnection attempts * CPP Client: Avoiding duplicate reconnection attempts * Fix for Unit Test
xiaoyaolanchong · Jul 11, 2017 · 4be5883 · 4be5883
1 parent f36b70c
commit 4be5883
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc
@@ -862,6 +862,7 @@ void ClientConnection::handleIncomingCommand() {
                     ProducersMap::iterator it = producers_.find(producerId);
                     if (it != producers_.end()) {
                         ProducerImplPtr producer = it->second.lock();
+                        producers_.erase(it);
                         lock.unlock();
 
                         if (producer) {
@@ -885,6 +886,7 @@ void ClientConnection::handleIncomingCommand() {
                     ConsumersMap::iterator it = consumers_.find(consumerId);
                     if (it != consumers_.end()) {
                         ConsumerImplPtr consumer = it->second.lock();
+                        consumers_.erase(it);
                         lock.unlock();
 
                         if (consumer) {
@@ -1124,6 +1126,10 @@ void ClientConnection::close() {
     state_ = Disconnected;
     boost::system::error_code err;
     socket_->close(err);
+    ConsumersMap consumers;
+    consumers.swap(consumers_);
+    ProducersMap producers;
+    producers.swap(producers_);
     lock.unlock();
 
     LOG_INFO(cnxString_ << "Connection closed");
@@ -1135,11 +1141,11 @@ void ClientConnection::close() {
     if (consumerStatsRequestTimer_) {
         consumerStatsRequestTimer_->cancel();
     }
-    for (ProducersMap::iterator it = producers_.begin(); it != producers_.end(); ++it) {
+    for (ProducersMap::iterator it = producers.begin(); it != producers.end(); ++it) {
         HandlerBase::handleDisconnection(ResultConnectError, shared_from_this(), it->second);
     }
 
-    for (ConsumersMap::iterator it = consumers_.begin(); it != consumers_.end(); ++it) {
+    for (ConsumersMap::iterator it = consumers.begin(); it != consumers.end(); ++it) {
         HandlerBase::handleDisconnection(ResultConnectError, shared_from_this(), it->second);
     }