Fix reconnection when heartbeat is missed

When a heartbeat is missing we call ensure_connection()
that runs a dummy method to trigger the reconnection
code in kombu. But also the code is triggered only if the
channel is None.

In case of the heartbeat threads we didn't reset the channel
before reconnecting, so the dummy method doesn't do anything.

This change sets the channel to None to ensure the connection
is reestablished before the dummy method is run.

Also it replaces the dummy method by checking the kombu connection
object. So we are sure the connection is reestablished.

Change-Id: I39f8cd23c5a5498e6f4c1aa3236ed27f3b5d7c9a
Closes-bug: #1493890

oslo_messaging/_drivers/impl_rabbit.py

@@ -581,7 +581,10 @@ class Connection(object):
         LOG.info(_LI("The broker has unblocked the connection"))
 
     def ensure_connection(self):
-        self.ensure(method=lambda: True)
+        # NOTE(sileht): we reset the channel and ensure
+        # the kombu underlying connection works
+        self._set_current_channel(None)
+        self.ensure(method=lambda: self.connection.connection)
 
     def ensure(self, method, retry=None,
                recoverable_error_callback=None, error_callback=None,
@@ -732,7 +735,6 @@ class Connection(object):
                 for tag, consumer in enumerate(self._consumers):
                     consumer.cancel(tag=tag)
             except recoverable_errors:
-                self._set_current_channel(None)
                 self.ensure_connection()
             self._consumers = []
 

oslo_messaging/tests/drivers/test_impl_rabbit.py

@@ -896,6 +896,8 @@ class RpcKombuHATestCase(test_utils.BaseTestCase):
         self.useFixture(mockpatch.Patch(
             'kombu.connection.Connection.connect',
             side_effect=self.kombu_connect))
+        self.useFixture(mockpatch.Patch(
+            'kombu.connection.Connection.connection'))
         self.useFixture(mockpatch.Patch(
             'kombu.connection.Connection.channel'))