Merge pull request #485 from bmbouter/455-close-qpid-when-session-closed

Causes the monitoring thread to exit if SessionClosed is raised.

kombu/tests/transport/test_qpid.py

@@ -1444,6 +1444,19 @@ def test_receivers_monitor_run_calls_monitor_receivers(
             self.monitor.run()
         mock_monitor_receivers.assert_called_once_with()
 
+    @patch(QPID_MODULE + '.SessionClosed', new=QpidException)
+    @patch.object(ReceiversMonitor, 'monitor_receivers')
+    @patch(QPID_MODULE + '.time.sleep')
+    def test_receivers_monitor_run_exits_on_session_closed(
+            self, mock_sleep, mock_monitor_receivers):
+        mock_monitor_receivers.side_effect = QpidException()
+        try:
+            self.monitor.run()
+        except Exception:
+            self.fail('No exception should be raised here')
+        mock_monitor_receivers.assert_called_once_with()
+        mock_sleep.has_calls([])
+
     @patch.object(Transport, 'connection_errors', new=(BreakOutException, ))
     @patch.object(ReceiversMonitor, 'monitor_receivers')
     @patch(QPID_MODULE + '.time.sleep')

kombu/transport/qpid.py

@@ -53,10 +53,12 @@
 try:
     from qpid.messaging.exceptions import ConnectionError, NotFound
     from qpid.messaging.exceptions import Empty as QpidEmpty
+    from qpid.messaging.exceptions import SessionClosed
 except ImportError:  # pragma: no cover
     ConnectionError = None
     NotFound = None
     QpidEmpty = None
+    SessionClosed = None
 
 try:
     import qpid
@@ -1338,6 +1340,10 @@ def run(self):
         non connection errors. This guards against unexpected exceptions
         which could cause this thread to exit unexpectedly.
 
+        A :class:`qpid.messaging.exceptions.SessionClosed` exception should
+        cause this thread to exit. This is a normal exit condition and the
+        thread is no longer needed.
+
         If a connection error occurs, the exception needs to be propagated
         to MainThread where the kombu exception handler can properly handle
         it. The exception is stored as saved_exception on the self._session
@@ -1351,6 +1357,8 @@ def run(self):
                 self._session.saved_exception = exc
                 os.write(self._w_fd, 'e')
                 break
+            except SessionClosed:
+                break
             except Exception as exc:
                 logger.error(exc, exc_info=1)
             time.sleep(10)