From 6ea680783a4ee9f4d9bcc34c2eb0b5f0eda6c98c Mon Sep 17 00:00:00 2001 From: Erik Olof Gunnar Andersson Date: Tue, 7 Nov 2023 04:33:24 -0800 Subject: [PATCH] Update heartbeat interval implementation [#127] --- amqpstorm/connection.py | 6 +++--- amqpstorm/heartbeat.py | 4 ++-- amqpstorm/tests/unit/test_heartbeat.py | 16 ++++++++-------- .../unit/uri_connection/test_uri_connection.py | 4 ++-- amqpstorm/uri_connection.py | 4 ++-- 5 files changed, 17 insertions(+), 17 deletions(-) diff --git a/amqpstorm/connection.py b/amqpstorm/connection.py index 34cd0643..2f4d9f20 100644 --- a/amqpstorm/connection.py +++ b/amqpstorm/connection.py @@ -22,7 +22,7 @@ LOGGER = logging.getLogger(__name__) -DEFAULT_HEARTBEAT_INTERVAL = 60 +DEFAULT_HEARTBEAT_TIMEOUT = 60 DEFAULT_SOCKET_TIMEOUT = 10 DEFAULT_VIRTUAL_HOST = '/' @@ -57,7 +57,7 @@ class Connection(Stateful): :param str password: Password :param int port: Server port :param str virtual_host: Virtual host - :param int heartbeat: RabbitMQ Heartbeat interval + :param int heartbeat: RabbitMQ Heartbeat timeout :param int,float timeout: Socket timeout :param bool ssl: Enable SSL :param dict ssl_options: SSL kwargs @@ -80,7 +80,7 @@ def __init__(self, hostname, username, password, port=5672, **kwargs): 'password': password, 'port': port, 'virtual_host': kwargs.get('virtual_host', DEFAULT_VIRTUAL_HOST), - 'heartbeat': kwargs.get('heartbeat', DEFAULT_HEARTBEAT_INTERVAL), + 'heartbeat': kwargs.get('heartbeat', DEFAULT_HEARTBEAT_TIMEOUT), 'timeout': kwargs.get('timeout', DEFAULT_SOCKET_TIMEOUT), 'ssl': kwargs.get('ssl', False), 'ssl_options': kwargs.get('ssl_options', {}), diff --git a/amqpstorm/heartbeat.py b/amqpstorm/heartbeat.py index 4366194e..e62fd001 100644 --- a/amqpstorm/heartbeat.py +++ b/amqpstorm/heartbeat.py @@ -11,7 +11,7 @@ class Heartbeat(object): """Internal Heartbeat handler.""" - def __init__(self, interval, send_heartbeat_impl, timer=threading.Timer): + def __init__(self, timeout, send_heartbeat_impl, timer=threading.Timer): self.send_heartbeat_impl = send_heartbeat_impl self.timer_impl = timer self._lock = threading.Lock() @@ -20,7 +20,7 @@ def __init__(self, interval, send_heartbeat_impl, timer=threading.Timer): self._exceptions = None self._reads_since_check = 0 self._writes_since_check = 0 - self._interval = interval + self._interval = None if timeout is None else max(timeout / 2, 0) self._threshold = 0 def register_read(self): diff --git a/amqpstorm/tests/unit/test_heartbeat.py b/amqpstorm/tests/unit/test_heartbeat.py index 7014d789..c96d5f15 100644 --- a/amqpstorm/tests/unit/test_heartbeat.py +++ b/amqpstorm/tests/unit/test_heartbeat.py @@ -43,15 +43,15 @@ def test_heartbeat_stop(self): self.assertFalse(heartbeat._running.is_set()) self.assertIsNone(heartbeat._timer) - def test_heartbeat_interval(self): + def test_heartbeat_timeout(self): heartbeat = Heartbeat(60, fake_function) - self.assertEqual(heartbeat._interval, 60) + self.assertEqual(heartbeat._interval, 30) self.assertEqual(heartbeat._threshold, 0) heartbeat = Heartbeat(360, fake_function) - self.assertEqual(heartbeat._interval, 360) + self.assertEqual(heartbeat._interval, 180) self.assertEqual(heartbeat._threshold, 0) def test_heartbeat_no_interval(self): @@ -229,7 +229,7 @@ def test_heartbeat_raise_exception(self): self.assertRaisesRegex( AMQPConnectionError, - 'Connection dead, no heartbeat or data received in >= 120s', + 'Connection dead, no heartbeat or data received in >= 60s', heartbeat._raise_or_append_exception ) @@ -237,7 +237,7 @@ def test_heartbeat_raise_exception(self): self.assertRaisesRegex( AMQPConnectionError, - 'Connection dead, no heartbeat or data received in >= 240', + 'Connection dead, no heartbeat or data received in >= 120s', heartbeat._raise_or_append_exception ) @@ -252,14 +252,14 @@ def check(exception): self.assertRaisesRegex( AMQPConnectionError, - 'Connection dead, no heartbeat or data received in >= 120s', + 'Connection dead, no heartbeat or data received in >= 60s', check, heartbeat._exceptions ) - heartbeat._interval = 120 + heartbeat._interval = 120 // 2 self.assertRaisesRegex( AMQPConnectionError, - 'Connection dead, no heartbeat or data received in >= 240', + 'Connection dead, no heartbeat or data received in >= 120s', check, heartbeat._exceptions ) diff --git a/amqpstorm/tests/unit/uri_connection/test_uri_connection.py b/amqpstorm/tests/unit/uri_connection/test_uri_connection.py index b621d6f8..24682d5f 100644 --- a/amqpstorm/tests/unit/uri_connection/test_uri_connection.py +++ b/amqpstorm/tests/unit/uri_connection/test_uri_connection.py @@ -1,7 +1,7 @@ import ssl from amqpstorm import UriConnection -from amqpstorm.connection import DEFAULT_HEARTBEAT_INTERVAL +from amqpstorm.connection import DEFAULT_HEARTBEAT_TIMEOUT from amqpstorm.connection import DEFAULT_SOCKET_TIMEOUT from amqpstorm.connection import DEFAULT_VIRTUAL_HOST from amqpstorm.tests.utility import TestFramework @@ -20,7 +20,7 @@ def test_uri_default(self): DEFAULT_VIRTUAL_HOST) self.assertEqual(connection.parameters['port'], 5672) self.assertEqual(connection.parameters['heartbeat'], - DEFAULT_HEARTBEAT_INTERVAL) + DEFAULT_HEARTBEAT_TIMEOUT) self.assertEqual(connection.parameters['timeout'], DEFAULT_SOCKET_TIMEOUT) self.assertFalse(connection.parameters['ssl']) diff --git a/amqpstorm/uri_connection.py b/amqpstorm/uri_connection.py index 6f89f374..e37ab712 100644 --- a/amqpstorm/uri_connection.py +++ b/amqpstorm/uri_connection.py @@ -6,7 +6,7 @@ from amqpstorm.compatibility import ssl from amqpstorm.compatibility import urlparse from amqpstorm.connection import Connection -from amqpstorm.connection import DEFAULT_HEARTBEAT_INTERVAL +from amqpstorm.connection import DEFAULT_HEARTBEAT_TIMEOUT from amqpstorm.connection import DEFAULT_SOCKET_TIMEOUT from amqpstorm.connection import DEFAULT_VIRTUAL_HOST from amqpstorm.exception import AMQPConnectionError @@ -83,7 +83,7 @@ def _parse_uri_options(self, parsed_uri, use_ssl=False, ssl_options=None): 'ssl': use_ssl, 'virtual_host': vhost, 'heartbeat': int(kwargs.pop('heartbeat', - [DEFAULT_HEARTBEAT_INTERVAL])[0]), + [DEFAULT_HEARTBEAT_TIMEOUT])[0]), 'timeout': int(kwargs.pop('timeout', [DEFAULT_SOCKET_TIMEOUT])[0]) }