QPID-1064: made qpid-config close the session/connection; added incoming.stop() to cancel incoming messages and join on the listener thread; made managementBroker.removeChannel use incoming.stop(); modified session.close to wait on _closed rather than on channel == None

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@656871 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 14 insertions, 5 deletions

diff --git a/python/commands/qpid-config b/python/commands/qpid-config
index 23a054f497..054ea39e94 100755
--- a/python/commands/qpid-config
+++ b/python/commands/qpid-config
@@ -108,6 +108,8 @@ class BrokerManager:
 
     def Disconnect (self):
         self.mclient.removeChannel (self.mchannel)
+        self.session.close(timeout=10)
+        self.conn.close(timeout=10)
 
     def Overview (self):
         self.ConnectToBroker ()
diff --git a/python/qpid/connection.py b/python/qpid/connection.py
index 8d0e115458..ce27a74489 100644
--- a/python/qpid/connection.py
+++ b/python/qpid/connection.py
@@ -102,7 +102,6 @@ class Connection(Assembler):
       if ssn is not None:
         ssn.channel = None
         ssn.closed()
-        notify(ssn.condition)
         return ssn
     finally:
       self.lock.release()
diff --git a/python/qpid/management.py b/python/qpid/management.py
index f831ee472e..6f68981e4c 100644
--- a/python/qpid/management.py
+++ b/python/qpid/management.py
@@ -129,8 +129,8 @@ class managementChannel:
 
   def shutdown (self):
     self.enabled = False
-    self.ssn.message_cancel (destination="tdest")
-    self.ssn.message_cancel (destination="rdest")
+    self.ssn.incoming("tdest").stop()
+    self.ssn.incoming("rdest").stop()
 
   def topicCb (self, msg):
     """ Receive messages via the topic queue on this channel. """
diff --git a/python/qpid/queue.py b/python/qpid/queue.py
index a8a5c0d9ad..830ea6060d 100644
--- a/python/qpid/queue.py
+++ b/python/qpid/queue.py
@@ -42,6 +42,9 @@ class Queue(BaseQueue):
   def close(self, error = None):
     self.error = error
     self.put(Queue.END)
+    if self.thread is not None:
+      self.thread.join()
+      self.thread = None
 
   def get(self, block = True, timeout = None):
     result = BaseQueue.get(self, block, timeout)
diff --git a/python/qpid/session.py b/python/qpid/session.py
index 6dc64b4f06..1b33a4a795 100644
--- a/python/qpid/session.py
+++ b/python/qpid/session.py
@@ -110,14 +110,13 @@ class Session(Invoker):
       self.channel.session_detach(self.name)
     finally:
       self.invoke_lock.release()
-    if not wait(self.condition, lambda: self.channel is None, timeout):
+    if not wait(self.condition, lambda: self._closed, timeout):
       raise Timeout()
 
   def closed(self):
     self.lock.acquire()
     try:
       if self._closed: return
-      self._closed = True
 
       error = self.error()
       for id in self.results:
@@ -127,6 +126,8 @@ class Session(Invoker):
 
       for q in self._incoming.values():
         q.close(error)
+
+      self._closed = True
       notify(self.condition)
     finally:
       self.lock.release()
@@ -344,6 +345,10 @@ class Incoming(Queue):
     for unit in self.session.credit_unit.values():
       self.session.message_flow(self.destination, unit, 0xFFFFFFFF)
 
+  def stop(self):
+    self.session.message_cancel(self.destination)
+    self.listen(None)
+
 class Delegate:
 
   def __init__(self, session):