reconnecting managements, publishers and consumers

Author: DanielePalaia

examples/reconnection/reconnection_example.py

@@ -129,7 +129,7 @@ def main() -> None:
                     publisher.publish(Message(body="test"))
             except ConnectionClosed:
                 print("publisher closing exception, resubmitting")
-                publisher = connection.publisher(addr)
+                # publisher = connection.publisher(addr)
                 continue
 
         print("closing publisher")
@@ -153,9 +153,6 @@ def main() -> None:
         except KeyboardInterrupt:
             pass
         except ConnectionClosed:
-            consumer = connection.consumer(
-                addr_queue, message_handler=MyMessageHandler()
-            )
             continue
         except Exception as e:
             print("consumer exited for exception " + str(e))
@@ -165,7 +162,6 @@ def main() -> None:
     print("cleanup")
     consumer.close()
 
-    management = connection.management()
     print("unbind")
     management.unbind(bind_name)
 

rabbitmq_amqp_python_client/connection.py

@@ -58,7 +58,7 @@ def __init__(
         self._addr: Optional[str] = uri
         self._addrs: Optional[list[str]] = uris
         self._conn: BlockingConnection
-        self._management: Management
+        self._managements: list[Management] = []
         self._reconnect = reconnect
         self._conf_ssl_context: Optional[SslConfigurationContext] = ssl_context
         self._ssl_domain = None
@@ -108,21 +108,20 @@ def dial(self) -> None:
                 on_disconnection_handler=self._on_disconnection,
             )
 
-        self._open()
+        # self._open()
         logger.debug("Connection to the server established")
 
-    def _open(self) -> None:
-        self._management = Management(self._conn)
-        self._management.open()
-
     def management(self) -> Management:
         """
         Get the management interface for this connection.
 
         Returns:
             Management: The management interface for performing administrative tasks
         """
-        return self._management
+        management = Management(self._conn)
+        management.open()
+        self._managements.append(management)
+        return management
 
     # closes the connection to the AMQP 1.0 server.
     def close(self) -> None:
@@ -137,6 +136,8 @@ def close(self) -> None:
                 publisher.close()
             for consumer in self._consumers:
                 consumer.close()
+            for management in self._managements:
+                management.close()
             self._conn.close()
         except Exception as e:
             logger.error(f"Error closing connection: {e}")
@@ -166,7 +167,7 @@ def publisher(self, destination: str = "") -> Publisher:
                 )
         publisher = Publisher(self._conn, destination)
         self._publishers.append(publisher)
-        return self._publishers[self._publishers.index(publisher)]
+        return publisher
 
     def consumer(
         self,
@@ -202,26 +203,26 @@ def consumer(
 
     def _on_disconnection(self) -> None:
 
-        print("disconnected")
-
         if self in self._connections:
             self._connections.remove(self)
 
-        print("reconnecting")
         self._conn = BlockingConnection(
             url=self._addr,
             urls=self._addrs,
             ssl_domain=self._ssl_domain,
             on_disconnection_handler=self._on_disconnection,
         )
-        self._open()
+
         self._connections.append(self)
 
-        for index, publisher in enumerate(self._publishers):
-            # publisher = self._publishers.pop(index)
-            # address = publisher.address
-            self._publishers.remove(publisher)
-            # self._publishers.insert(index, Publisher(self._conn, address))
+        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, consumer in enumerate(self._consumers):
-            self._consumers.remove(consumer)
+            # Update the broken connection and sender in the consumer
+            self._consumers[i].update_connection(self._conn)

rabbitmq_amqp_python_client/consumer.py

@@ -65,6 +65,10 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._receiver = self._create_receiver(self._addr)
 
+    def update_connection(self, conn: BlockingConnection) -> None:
+        self._conn = conn
+        self._receiver = self._create_receiver(self._addr)
+
     def _set_consumers_list(self, consumers: []) -> None:  # type: ignore
         self._consumers = consumers
 

rabbitmq_amqp_python_client/management.py

@@ -53,6 +53,13 @@ def __init__(self, conn: BlockingConnection):
         self._receiver: Optional[BlockingReceiver] = None
         self._conn = conn
 
+    def update_connection(self, conn: BlockingConnection) -> None:
+        self._conn = conn
+        self._sender = self._create_sender(CommonValues.management_node_address.value)
+        self._receiver = self._create_receiver(
+            CommonValues.management_node_address.value,
+        )
+
     def open(self) -> None:
         """
         Open the management connection by creating sender and receiver.

rabbitmq_amqp_python_client/publisher.py

@@ -46,6 +46,10 @@ def __init__(self, conn: BlockingConnection, addr: str = ""):
         self._publishers: list[Publisher] = []
         self._open()
 
+    def update_connection(self, conn: BlockingConnection) -> None:
+        self._conn = conn
+        self._sender = self._create_sender(self._addr)
+
     def _open(self) -> None:
         if self._sender is None:
             logger.debug("Creating Sender")

tests/test_connection.py

@@ -90,9 +90,11 @@ def test_connection_reconnection() -> None:
     # delay
     time.sleep(5)
     # simulate a disconnection
-    delete_all_connections()
     # raise a reconnection
     management = connection.management()
+
+    delete_all_connections()
+
     stream_name = "test_stream_info_with_validation"
     queue_specification = StreamSpecification(
         name=stream_name,
@@ -104,7 +106,6 @@ def test_connection_reconnection() -> None:
         disconnected = True
 
     # check that we reconnected
-    management = connection.management()
     management.declare_queue(queue_specification)
     management.delete_queue(stream_name)
     management.close()

tests/test_publisher.py

@@ -291,9 +291,6 @@ def test_disconnection_reconnection() -> None:
 
             except ConnectionClosed:
                 disconnected = True
-                publisher = connection_test.publisher(
-                    destination=AddressHelper.queue_address(queue_name)
-                )
                 continue
 
             except Exception:
@@ -303,14 +300,6 @@ def test_disconnection_reconnection() -> None:
 
     publisher.close()
 
-    # cleanup, we need to create a new connection as the previous one
-    # was closed by the test
-
-    connection_test = environment.connection()
-    connection_test.dial()
-
-    management = connection_test.management()
-
     # purge the queue and check number of published messages
     message_purged = management.purge_queue(queue_name)
 
@@ -335,14 +324,10 @@ def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
     management = connection.management()
     management.declare_queue(queue_specification)
 
-    print("before creating publisher")
-
     publisher = connection.publisher(
         destination=AddressHelper.queue_address(stream_name)
     )
 
-    print("after creating publisher")
-
     for i in range(messages_to_send):
 
         publisher.publish(Message(body="test"))