diff options
| -rw-r--r-- | oslo_messaging/_drivers/impl_rabbit.py | 8 | ||||
| -rw-r--r-- | oslo_messaging/tests/drivers/test_impl_rabbit.py | 12 |
2 files changed, 6 insertions, 14 deletions
diff --git a/oslo_messaging/_drivers/impl_rabbit.py b/oslo_messaging/_drivers/impl_rabbit.py index 8c696cb..75041cd 100644 --- a/oslo_messaging/_drivers/impl_rabbit.py +++ b/oslo_messaging/_drivers/impl_rabbit.py @@ -231,6 +231,9 @@ class Consumer(object): consumer_tag=six.text_type(tag), nowait=self.nowait) + def cancel(self, tag): + self.queue.cancel(six.text_type(tag)) + def _callback(self, message): """Call callback with deserialized message. @@ -689,11 +692,12 @@ class Connection(object): with self._connection_lock: try: - self._set_current_channel(self.connection.channel()) + for tag, consumer in enumerate(self._consumers): + consumer.cancel(tag=tag) except recoverable_errors: self._set_current_channel(None) self.ensure_connection() - self._consumers = [] + self._consumers = [] def _heartbeat_supported_and_enabled(self): if self.driver_conf.heartbeat_timeout_threshold <= 0: diff --git a/oslo_messaging/tests/drivers/test_impl_rabbit.py b/oslo_messaging/tests/drivers/test_impl_rabbit.py index 1e942a9..48ffffc 100644 --- a/oslo_messaging/tests/drivers/test_impl_rabbit.py +++ b/oslo_messaging/tests/drivers/test_impl_rabbit.py @@ -257,18 +257,6 @@ class TestRabbitConsume(test_utils.BaseTestCase): self.assertEqual(0, int(deadline - time.time())) - def test_connection_reset_always_succeed(self): - transport = oslo_messaging.get_transport(self.conf, - 'kombu+memory:////') - self.addCleanup(transport.cleanup) - channel = mock.Mock() - with transport._driver._get_connection(amqp.PURPOSE_LISTEN) as conn: - conn.connection.connection.recoverable_channel_errors = (IOError,) - with mock.patch.object(conn.connection.connection, 'channel', - side_effect=[IOError, IOError, channel]): - conn.connection.reset() - self.assertEqual(channel, conn.connection.channel) - def test_connection_ack_have_disconnected_kombu_connection(self): transport = oslo_messaging.get_transport(self.conf, 'kombu+memory:////') |