implement BackOff and MaxReconnectAttempts

Author: DanielePalaia

examples/reconnection/reconnection_example.py

@@ -10,13 +10,17 @@
     ExchangeToQueueBindingSpecification,
     Message,
     QuorumQueueSpecification,
+    RecoveryConfiguration,
 )
 
 # here we keep track of the objects we need to reconnect
 MESSAGES_TO_PUBLISH = 50000
 
 
-environment = Environment(uri="amqp://guest:guest@localhost:5672/", reconnect=True)
+environment = Environment(
+    uri="amqp://guest:guest@localhost:5672/",
+    recovery_configuration=RecoveryConfiguration(active_recovery=True),
+)
 
 
 class MyMessageHandler(AMQPMessagingHandler):

rabbitmq_amqp_python_client/__init__.py

@@ -11,6 +11,7 @@
     ExchangeToExchangeBindingSpecification,
     ExchangeToQueueBindingSpecification,
     OffsetSpecification,
+    RecoveryConfiguration,
     StreamOptions,
 )
 from .environment import Environment
@@ -87,4 +88,5 @@
     "OutcomeState",
     "Environment",
     "ExchangeCustomSpecification",
+    "RecoveryConfiguration",
 ]

rabbitmq_amqp_python_client/connection.py

@@ -1,4 +1,7 @@
 import logging
+import random
+import time
+from datetime import timedelta
 from typing import (
     Annotated,
     Callable,
@@ -11,10 +14,11 @@
 
 from .address_helper import validate_address
 from .consumer import Consumer
-from .entities import StreamOptions
+from .entities import RecoveryConfiguration, StreamOptions
 from .exceptions import ArgumentOutOfRangeException
 from .management import Management
 from .publisher import Publisher
+from .qpid.proton._exceptions import ConnectionException
 from .qpid.proton._handlers import MessagingHandler
 from .qpid.proton._transport import SSLDomain
 from .qpid.proton.utils import BlockingConnection
@@ -53,7 +57,7 @@ def __init__(
         ssl_context: Union[
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = None,
-        reconnect: bool = False,
+        recovery_configuration: Optional[RecoveryConfiguration] = None,
     ):
         """
          Initialize a new Connection instance.
@@ -80,7 +84,7 @@ def __init__(
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = ssl_context
         self._managements: list[Management] = []
-        self._reconnect = reconnect
+        self._recovery_configuration = recovery_configuration
         self._ssl_domain = None
         self._connections = []  # type: ignore
         self._index: int = -1
@@ -144,7 +148,10 @@ def dial(self) -> None:
                     password,
                 )
 
-        if self._reconnect is False:
+        if (
+            self._recovery_configuration is None
+            or self._recovery_configuration.active_recovery is False
+        ):
             self._conn = BlockingConnection(
                 url=self._addr,
                 urls=self._addrs,
@@ -274,26 +281,53 @@ def _on_disconnection(self) -> None:
         if self in self._connections:
             self._connections.remove(self)
 
-        self._conn = BlockingConnection(
-            url=self._addr,
-            urls=self._addrs,
-            ssl_domain=self._ssl_domain,
-            on_disconnection_handler=self._on_disconnection,
-        )
+        base_delay = self._recovery_configuration.back_off_reconnect_interval  # type: ignore
+        max_delay = timedelta(minutes=1)
+
+        for attempt in range(self._recovery_configuration.MaxReconnectAttempts):  # type: ignore
+
+            jitter = timedelta(milliseconds=random.randint(0, 500))
+            delay = base_delay + jitter
+
+            if delay > max_delay:
+                delay = max_delay
+
+            time.sleep(delay.total_seconds())
 
-        self._connections.append(self)
+            try:
+                self._conn = BlockingConnection(
+                    url=self._addr,
+                    urls=self._addrs,
+                    ssl_domain=self._ssl_domain,
+                    on_disconnection_handler=self._on_disconnection,
+                )
+
+                self._connections.append(self)
+
+                for i, management in enumerate(self._managements):
+                    # Update the broken connection and sender in the management
+                    self._managements[i]._update_connection(self._conn)
 
-        for i, management in enumerate(self._managements):
-            # Update the broken connection and sender in the management
-            self._managements[i]._update_connection(self._conn)
+                for i, publisher in enumerate(self._publishers):
+                    # Update the broken connection and sender in the publisher
+                    self._publishers[i]._update_connection(self._conn)
 
-        for i, publisher in enumerate(self._publishers):
-            # Update the broken connection and sender in the publisher
-            self._publishers[i]._update_connection(self._conn)
+                for i, consumer in enumerate(self._consumers):
+                    # Update the broken connection and sender in the consumer
+                    self._consumers[i]._update_connection(self._conn)
+
+            except ConnectionException as e:
+                base_delay *= 2
+                logger.error(
+                    "Reconnection attempt failed",
+                    "attempt",
+                    attempt,
+                    "error",
+                    str(e),
+                )
+                continue
 
-        for i, consumer in enumerate(self._consumers):
-            # Update the broken connection and sender in the consumer
-            self._consumers[i]._update_connection(self._conn)
+            break
 
     @property
     def active_producers(self) -> int:

rabbitmq_amqp_python_client/entities.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass, field
+from datetime import timedelta
 from enum import Enum
 from typing import Any, Dict, Optional, Union
 
@@ -208,3 +209,27 @@ def filter_set(self) -> Dict[symbol, Described]:
             Dict[symbol, Described]: The current filter set configuration
         """
         return self._filter_set
+
+
+@dataclass
+class RecoveryConfiguration:
+    """
+    Configuration options for automatic reconnection.
+
+    This dataclass contains parameters to manage automatic reconnection
+
+    Attributes:
+    active_recovery: Define if the recovery is activated. If is not activated the connection will not try to reconnect
+    back_off_reconnect_interval: the time to wait before trying to createSender after a connection is closed.
+                time will be increased exponentially with each attempt.
+                Default is 5 seconds, each attempt will double the time.
+                The minimum value is 1 second. Avoid setting a value low values since it can cause a high
+                number of reconnection attempts.
+        MaxReconnectAttempts: 		MaxReconnectAttempts The maximum number of reconnection attempts.
+                Default is 5.
+                The minimum value is 1.
+    """
+
+    active_recovery: bool = True
+    back_off_reconnect_interval: timedelta = timedelta(0.5)
+    MaxReconnectAttempts: int = 5

rabbitmq_amqp_python_client/environment.py

@@ -9,6 +9,7 @@
 )
 
 from .connection import Connection
+from .entities import RecoveryConfiguration
 from .ssl_configuration import (
     PosixSslConfigurationContext,
     WinSslConfigurationContext,
@@ -40,7 +41,7 @@ def __init__(
         ssl_context: Union[
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = None,
-        reconnect: bool = False,
+        recovery_configuration: Optional[RecoveryConfiguration] = None,
     ):
         """
         Initialize a new Environment instance.
@@ -63,7 +64,7 @@ def __init__(
         self._uri = uri
         self._uris = uris
         self._ssl_context = ssl_context
-        self._reconnect = reconnect
+        self._recovery_configuration = recovery_configuration
         self._connections: list[Connection] = []
 
     def connection(
@@ -85,7 +86,7 @@ def connection(
             uri=self._uri,
             uris=self._uris,
             ssl_context=self._ssl_context,
-            reconnect=self._reconnect,
+            recovery_configuration=self._recovery_configuration,
         )
         logger.debug("Environment: Creating and returning a new connection")
         self._connections.append(connection)

rabbitmq_amqp_python_client/qpid/proton/_tracing.py

@@ -32,8 +32,10 @@
 
 import proton
 from proton import Sender as ProtonSender
-from proton.handlers import IncomingMessageHandler as ProtonIncomingMessageHandler
-from proton.handlers import OutgoingMessageHandler as ProtonOutgoingMessageHandler
+from proton.handlers import \
+    IncomingMessageHandler as ProtonIncomingMessageHandler
+from proton.handlers import \
+    OutgoingMessageHandler as ProtonOutgoingMessageHandler
 
 _tracer = None
 _trace_key = proton.symbol("x-opt-qpid-tracestate")

rabbitmq_amqp_python_client/qpid/proton/_transport.py

@@ -139,7 +139,8 @@
 
 if TYPE_CHECKING:
     from ._condition import Condition
-    from ._endpoints import Connection  # would produce circular import
+    from ._endpoints import \
+        Connection  # would produce circular import
 
 
 class TraceAdapter:

tests/conftest.py

@@ -12,6 +12,7 @@
     PKCS12Store,
     PosixClientCert,
     PosixSslConfigurationContext,
+    RecoveryConfiguration,
     WinClientCert,
     WinSslConfigurationContext,
     symbol,
@@ -47,6 +48,21 @@ def connection(pytestconfig):
         environment.close()
 
 
+@pytest.fixture()
+def connection_with_reconnect(pytestconfig):
+    environment = Environment(
+        uri="amqp://guest:guest@localhost:5672/",
+        recovery_configuration=RecoveryConfiguration(active_recovery=True),
+    )
+    connection = environment.connection()
+    connection.dial()
+    try:
+        yield connection
+
+    finally:
+        environment.close()
+
+
 @pytest.fixture()
 def ssl_context(pytestconfig):
     if sys.platform == "win32":

tests/test_connection.py

@@ -3,6 +3,7 @@
 from rabbitmq_amqp_python_client import (
     ConnectionClosed,
     Environment,
+    RecoveryConfiguration,
     StreamSpecification,
 )
 
@@ -11,7 +12,6 @@
 
 def on_disconnected():
 
-    print("disconnected")
     global disconnected
     disconnected = True
 
@@ -73,7 +73,10 @@ def test_connection_reconnection() -> None:
 
     disconnected = False
 
-    environment = Environment("amqp://guest:guest@localhost:5672/", reconnect=True)
+    environment = Environment(
+        "amqp://guest:guest@localhost:5672/",
+        recovery_configuration=RecoveryConfiguration(active_recovery=True),
+    )
 
     connection = environment.connection()
     connection.dial()

tests/test_publisher.py

@@ -9,6 +9,7 @@
     Message,
     OutcomeState,
     QuorumQueueSpecification,
+    RecoveryConfiguration,
     StreamSpecification,
     ValidationCodeException,
 )
@@ -262,7 +263,10 @@ def test_disconnection_reconnection() -> None:
     disconnected = False
     generic_exception_raised = False
 
-    environment = Environment("amqp://guest:guest@localhost:5672/", reconnect=True)
+    environment = Environment(
+        "amqp://guest:guest@localhost:5672/",
+        recovery_configuration=RecoveryConfiguration(active_recovery=True),
+    )
 
     connection_test = environment.connection()