rabbit: heartbeat implementation

AMQP offers a heartbeat feature to ensure that the application layer
promptly finds out about disrupted connections (and also completely
unresponsive peers). If the client requests heartbeats on connection, rabbit
server will regularly send messages to each connections with the expectation of
a response.

To acheive this, each driver connection object spawn a thread that
send/retrieve heartbeat packets exchanged between the server and the
client.

To protect the concurrency access to the kombu connection between the
driver and this thread use a lock that always prioritize the
heartbeat thread. So when the heartbeat thread wakes up it will acquire the
lock quickly, to ensure we have no heartbeat starvation when the driver
sends a lot of messages.

Also when we are polling the broker, the lock can be held for a long
time by the 'consume' method, so this one does the heartbeat stuffs itself.

DocImpact: 2 new configuration options for Rabbit driver

Co-Authored-By: Oleksii Zamiatin <ozamiatin@mirantis.com>
Co-Authored-By: Ilya Pekelny <ipekelny@mirantis.com>

Related-Bug: #1371723
Closes-Bug: #856764

Change-Id: I1d3a635f3853bc13ffc14034468f1ac6262c11a3
(cherry picked from commit b9e134d7e955b9180482d2f7c8844501c750adf6)

1 files changed, 5 insertions, 2 deletions

diff --git a/tests/drivers/test_impl_qpid.py b/tests/drivers/test_impl_qpid.py
index d8cd1e7..2c2c0a5 100644
--- a/tests/drivers/test_impl_qpid.py
+++ b/tests/drivers/test_impl_qpid.py
@@ -27,6 +27,7 @@ import testscenarios
 import testtools
 
 from oslo import messaging
+from oslo_messaging._drivers import amqp
 from oslo_messaging._drivers import impl_qpid as qpid_driver
 from oslo_messaging.tests import utils as test_utils
 
@@ -564,7 +565,8 @@ class TestQpidReconnectOrder(test_utils.BaseTestCase):
         with mock.patch('qpid.messaging.Connection') as conn_mock:
             # starting from the first broker in the list
             url = messaging.TransportURL.parse(self.conf, None)
-            connection = qpid_driver.Connection(self.conf, url)
+            connection = qpid_driver.Connection(self.conf, url,
+                                                amqp.PURPOSE_SEND)
 
             # reconnect will advance to the next broker, one broker per
             # attempt, and then wrap to the start of the list once the end is
@@ -806,7 +808,8 @@ class QPidHATestCase(test_utils.BaseTestCase):
 
         # starting from the first broker in the list
         url = messaging.TransportURL.parse(self.conf, None)
-        self.connection = qpid_driver.Connection(self.conf, url)
+        self.connection = qpid_driver.Connection(self.conf, url,
+                                                 amqp.PURPOSE_SEND)
         self.addCleanup(self.connection.close)
 
         self.info.update({'attempt': 0,