From 9a3d8ad1695d1b7f2cb42d77c6106245c521566c Mon Sep 17 00:00:00 2001 From: "Rafael H. Schloming" Date: Thu, 15 May 2008 22:15:33 +0000 Subject: 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 --- python/qpid/session.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) (limited to 'python/qpid/session.py') 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): -- cgit v1.2.1