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

1 files changed, 7 insertions, 2 deletions

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):