diff options
Diffstat (limited to 'python/qpid')
| -rw-r--r-- | python/qpid/connection.py | 1 | ||||
| -rw-r--r-- | python/qpid/management.py | 4 | ||||
| -rw-r--r-- | python/qpid/queue.py | 3 | ||||
| -rw-r--r-- | python/qpid/session.py | 9 |
4 files changed, 12 insertions, 5 deletions
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): |