1 files changed, 26 insertions, 2 deletions

diff --git a/cpp/src/qpid/console/Broker.cpp b/cpp/src/qpid/console/Broker.cpp
index 789b90eaaf..4f90afd39a 100644
--- a/cpp/src/qpid/console/Broker.cpp
+++ b/cpp/src/qpid/console/Broker.cpp
@@ -57,6 +57,8 @@ Broker::Broker(SessionManager& sm, ConnectionSettings& settings) :
 
 Broker::~Broker()
 {
+    connThreadBody.shutdown();
+    connThread.join();
 }
 
 string Broker::getUrl() const
@@ -184,6 +186,8 @@ void Broker::ConnectionThread::run()
             subscriptions->setFlowControl(dest, FlowControl::unlimited());
             {
                 Mutex::ScopedLock _lock(connLock);
+                if (shuttingDown)
+                    return;
                 operational = true;
                 broker.resetAgents();
                 broker.connected = true;
@@ -199,16 +203,26 @@ void Broker::ConnectionThread::run()
                 broker.sessionManager.handleBrokerDisconnect(&broker);
             }
             delay = delayMin;
+            connection.close();
             delete subscriptions;
             subscriptions = 0;
-            session.close();
         } catch (std::exception &e) {
             QPID_LOG(debug, "  outer exception: " << e.what());
             if (delay < delayMax)
                 delay *= delayFactor;
         }
 
-        ::sleep(delay);
+            {
+                Mutex::ScopedLock _lock(connLock);
+                if (shuttingDown)
+                    return;
+                {
+                    Mutex::ScopedUnlock _unlock(connLock);
+                    ::sleep(delay);
+                }
+                if (shuttingDown)
+                    return;
+            }
     }
 }
 
@@ -253,6 +267,16 @@ void Broker::ConnectionThread::bindExchange(const std::string& exchange, const s
                           arg::bindingKey=key);
 }
 
+void Broker::ConnectionThread::shutdown()
+{
+    {
+        Mutex::ScopedLock _lock(connLock);
+        shuttingDown = true;
+    }
+    if (subscriptions)
+        subscriptions->stop();
+}
+
 void Broker::waitForStable()
 {
     Mutex::ScopedLock l(lock);