multinode implementation

Author: DanielePalaia

examples/getting_started/main.py

@@ -59,7 +59,7 @@ def on_link_closed(self, event: Event) -> None:
 
 
 def create_connection() -> Connection:
-    connection = Connection("amqps://guest:guest@localhost:5672/")
+    connection = Connection("amqp://guest:guest@localhost:5672/")
     # in case of SSL enablement
     # ca_cert_file = ".ci/certs/ca_certificate.pem"
     # client_cert = ".ci/certs/client_certificate.pem"

examples/getting_started/reconnection_example.py

@@ -21,7 +21,9 @@
 consumer = None
 
 
-def on_disconnected():
+# disconnection callback
+# here you can cleanup or reconnect
+def on_disconnection():
 
     print("disconnected")
     exchange_name = "test-exchange"
@@ -36,7 +38,8 @@ def on_disconnected():
     addr = AddressHelper.exchange_address(exchange_name, routing_key)
     addr_queue = AddressHelper.queue_address(queue_name)
 
-    connection = create_connection()
+    if connection is not None:
+        connection = create_connection()
     if management is not None:
         management = connection.management()
     if publisher is not None:
@@ -93,12 +96,13 @@ def on_link_closed(self, event: Event) -> None:
 def create_connection() -> Connection:
     # for multinode specify a list of urls and fill the field urls of Connection instead of url
     # urls = [
-    #    "amqp://ha_tls-rabbit_node0-1:5602/",
-    #    "amqp://ha_tls-rabbit_node0-2:5602/",
-    #    "amqp://ha_tls-rabbit_node0-3:5602/",
+    #    "amqp://ha_tls-rabbit_node0-1:5682/",
+    #    "amqp://ha_tls-rabbit_node1-1:5692/",
+    #    "amqp://ha_tls-rabbit_node2-1:5602/",
     # ]
+    # connection = Connection(urls=urls, on_disconnection_handler=on_disconnected)
     connection = Connection(
-        "amqp://guest:guest@localhost:5672/", on_disconnection_handler=on_disconnected
+        url="amqp://guest:guest@localhost:5672/", on_disconnection_handler=on_disconnection
     )
     connection.dial()
 

rabbitmq_amqp_python_client/qpid/proton/_utils.py

@@ -431,37 +431,46 @@ def __init__(
         on_disconnection_handler: Optional[CB] = None,
         **kwargs
     ) -> None:
-        self.disconnected = False
-        self.timeout = timeout or 60
-        self.container = container or Container()
-        self.container.timeout = self.timeout
-        self.container.start()
-        self.conn = None
-        self.closing = False
-        self._on_disconnection_handler = on_disconnection_handler
+
         # Preserve previous behaviour if neither reconnect nor urls are supplied
-        if url is not None and urls is None and reconnect is None:
-            reconnect = False
-            url = Url(url).defaults()
-        failed = True
-        try:
+        if urls is None:
+            urls = []
+            urls.append(url)
+
+        # multinode reimplementation (default one wasn't working properly)
+        attempt = 0
+        for i in range(len(urls)):
+            attempt = attempt + 1
+            self.disconnected = False
+            self.timeout = timeout or 60
+            self.container = container or Container()
+            self.container.timeout = self.timeout
+            self.container.start()
+            self.conn = None
+            self.closing = False
+            self._on_disconnection_handler = on_disconnection_handler
+
+            url_it = urls[i]
             self.conn = self.container.connect(
-                url=url,
+                url=Url(url_it).defaults(),
                 handler=self,
                 ssl_domain=ssl_domain,
-                reconnect=reconnect,
+                reconnect=False,
                 heartbeat=heartbeat,
-                urls=urls,
+                urls=None,
                 **kwargs
             )
-            self.wait(
-                lambda: not (self.conn.state & Endpoint.REMOTE_UNINIT),
-                msg="Opening connection",
-            )
-            failed = False
-        finally:
-            if failed and self.conn:
-                self.close()
+            try:
+                self.wait(
+                    lambda: not (self.conn.state & Endpoint.REMOTE_UNINIT),
+                    msg="Opening connection",
+                )
+
+            except ConnectionException:
+                self.conn.close()
+                if attempt == len(urls):
+                    raise
+                continue
 
     def create_sender(
         self,