Introducing Environment

Author: DanielePalaia

README.md

@@ -38,12 +38,13 @@ poetry run python ./examples/getting_started/main.py
 
 ### Creating a connection
 
-A connection to the RabbitMQ AMQP 1.0 server can be established using the Connection object.
+A connection to the RabbitMQ AMQP 1.0 server can be established using the Environment object.
 
 For example:
 
 ```python
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+    environment = Environment()
+    connection = environment.connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
 ```
 

examples/getting_started/basic_example.py

@@ -11,6 +11,7 @@
     ExchangeSpecification,
     Message,
     QuorumQueueSpecification,
+    Environment,
 )
 
 MESSAGES_TO_PUBLISH = 100
@@ -61,8 +62,19 @@ def on_link_closed(self, event: Event) -> None:
         print("link closed")
 
 
-def create_connection() -> Connection:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+def create_connection(environment: Environment) -> Connection:
+    connection = environment.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"
+    # client_key = ".ci/certs/client_key.pem"
+    # connection = Connection(
+    #    "amqps://guest:guest@localhost:5671/",
+    #    ssl_context=SslConfigurationContext(
+    #        ca_cert=ca_cert_file,
+    #        client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
+    #    ),
+    # )
     connection.dial()
 
     return connection
@@ -75,7 +87,8 @@ def main() -> None:
     routing_key = "routing-key"
 
     print("connection to amqp server")
-    connection = create_connection()
+    environment = Environment()
+    connection = create_connection(environment)
 
     management = connection.management()
 
@@ -150,7 +163,7 @@ def main() -> None:
     print("closing connections")
     management.close()
     print("after management closing")
-    connection.close()
+    environment.close()
     print("after connection closing")
 
 

examples/getting_started/example_with_streams.py

@@ -9,6 +9,7 @@
     OffsetSpecification,
     StreamOptions,
     StreamSpecification,
+    Environment,
 )
 
 MESSAGES_TO_PUBLISH = 100
@@ -65,8 +66,19 @@ def on_link_closed(self, event: Event) -> None:
         print("link closed")
 
 
-def create_connection() -> Connection:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+def create_connection(environment: Environment) -> Connection:
+    connection = environment.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"
+    # client_key = ".ci/certs/client_key.pem"
+    # connection = Connection(
+    #    "amqps://guest:guest@localhost:5671/",
+    #    ssl_context=SslConfigurationContext(
+    #        ca_cert=ca_cert_file,
+    #        client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
+    #    ),
+    # )
     connection.dial()
 
     return connection
@@ -76,15 +88,16 @@ def main() -> None:
     queue_name = "example-queue"
 
     print("connection to amqp server")
-    connection = create_connection()
+    environment = Environment()
+    connection = create_connection(environment)
 
     management = connection.management()
 
     management.declare_queue(StreamSpecification(name=queue_name))
 
     addr_queue = AddressHelper.queue_address(queue_name)
 
-    consumer_connection = create_connection()
+    consumer_connection = create_connection(environment)
 
     stream_filter_options = StreamOptions()
     # can be first, last, next or an offset long
@@ -135,7 +148,7 @@ def main() -> None:
     print("closing connections")
     management.close()
     print("after management closing")
-    connection.close()
+    environment.close()
     print("after connection closing")
 
 

examples/getting_started/reconnection_example.py

@@ -18,9 +18,10 @@
     Message,
     Publisher,
     QuorumQueueSpecification,
+    Environment,
 )
 
-
+environment = Environment()
 # here we keep track of the objects we need to reconnect
 @dataclass
 class ConnectionConfiguration:
@@ -118,8 +119,9 @@ def create_connection() -> Connection:
     #    "amqp://ha_tls-rabbit_node2-1:5602/",
     # ]
     # connection = Connection(uris=uris, on_disconnection_handler=on_disconnected)
-    connection = Connection(
-        uri="amqp://guest:guest@localhost:5672/",
+
+    connection = environment.connection(
+        url="amqp://guest:guest@localhost:5672/",
         on_disconnection_handler=on_disconnection,
     )
     connection.dial()
@@ -242,7 +244,7 @@ def main() -> None:
     print("closing connections")
     connection_configuration.management.close()
     print("after management closing")
-    connection_configuration.connection.close()
+    environment.close()
     print("after connection closing")
 
 

examples/getting_started/tls_example.py

@@ -12,6 +12,7 @@
     Message,
     QuorumQueueSpecification,
     SslConfigurationContext,
+    Environment,
 )
 
 messages_to_publish = 100
@@ -62,12 +63,12 @@ def on_link_closed(self, event: Event) -> None:
         print("link closed")
 
 
-def create_connection() -> Connection:
+def create_connection(environment: Environment) -> Connection:
     # in case of SSL enablement
     ca_cert_file = ".ci/certs/ca_certificate.pem"
     client_cert = ".ci/certs/client_certificate.pem"
     client_key = ".ci/certs/client_key.pem"
-    connection = Connection(
+    connection = environment.connection(
         "amqps://guest:guest@localhost:5671/",
         ssl_context=SslConfigurationContext(
             ca_cert=ca_cert_file,
@@ -84,9 +85,10 @@ def main() -> None:
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
+    environment = Environment()
 
     print("connection to amqp server")
-    connection = create_connection()
+    connection = create_connection(environment)
 
     management = connection.management()
 
@@ -160,7 +162,7 @@ def main() -> None:
     print("closing connections")
     management.close()
     print("after management closing")
-    connection.close()
+    environment.close()
     print("after connection closing")
 
 

rabbitmq_amqp_python_client/__init__.py

@@ -11,6 +11,7 @@
     OffsetSpecification,
     StreamOptions,
 )
+from .environment import Environment
 from .exceptions import ArgumentOutOfRangeException
 from .management import Management
 from .publisher import Publisher
@@ -66,4 +67,5 @@
     "StreamOptions",
     "OffsetSpecification",
     "Disposition",
+    "Environment",
 ]

rabbitmq_amqp_python_client/connection.py

@@ -72,7 +72,7 @@ def management(self) -> Management:
         return self._management
 
     # closes the connection to the AMQP 1.0 server.
-    def close(self) -> None:
+    def _close(self) -> None:
         logger.debug("Closing connection")
         self._conn.close()
 

rabbitmq_amqp_python_client/environment.py

@@ -0,0 +1,37 @@
+from typing import Annotated, Callable, Optional, TypeVar
+
+from .connection import Connection
+from .ssl_configuration import SslConfigurationContext
+
+MT = TypeVar("MT")
+CB = Annotated[Callable[[MT], None], "Message callback type"]
+
+
+class Environment:
+
+    def __init__(self):  # type: ignore
+
+        self._connections = []
+
+    def connection(
+        self,
+        # single-node mode
+        url: Optional[str] = None,
+        # multi-node mode
+        urls: Optional[list[str]] = None,
+        ssl_context: Optional[SslConfigurationContext] = None,
+        on_disconnection_handler: Optional[CB] = None,  # type: ignore
+    ) -> Connection:
+        connection = Connection(
+            url=url,
+            urls=urls,
+            ssl_context=ssl_context,
+            on_disconnection_handler=on_disconnection_handler,
+        )
+
+        self._connections.append(connection)
+        return connection
+
+    def close(self) -> None:
+        for connection in self._connections:
+            connection._close()

tests/conftest.py

@@ -6,30 +6,42 @@
     AddressHelper,
     AMQPMessagingHandler,
     ClientCert,
-    Connection,
+    Environment,
     Event,
     SslConfigurationContext,
     symbol,
 )
 
 
+@pytest.fixture()
+def environment(pytestconfig):
+    environment = Environment()
+    try:
+        yield environment
+
+    finally:
+        environment.close()
+
+
 @pytest.fixture()
 def connection(pytestconfig):
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+    environment = Environment()
+    connection = environment.connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
     try:
         yield connection
 
     finally:
-        connection.close()
+        environment.close()
 
 
 @pytest.fixture()
 def connection_ssl(pytestconfig):
+    environment = Environment()
     ca_cert_file = ".ci/certs/ca_certificate.pem"
     client_cert = ".ci/certs/client_certificate.pem"
     client_key = ".ci/certs/client_key.pem"
-    connection = Connection(
+    connection = environment.connection(
         "amqps://guest:guest@localhost:5671/",
         ssl_context=SslConfigurationContext(
             ca_cert=ca_cert_file,
@@ -41,25 +53,26 @@ def connection_ssl(pytestconfig):
         yield connection
 
     finally:
-        connection.close()
+        environment.close()
 
 
 @pytest.fixture()
 def management(pytestconfig):
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+    environment = Environment()
+    connection = environment.connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
     try:
         management = connection.management()
         yield management
 
     finally:
-        management.close()
-        connection.close()
+        environment.close()
 
 
 @pytest.fixture()
 def consumer(pytestconfig):
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+    environment = Environment()
+    connection = environment.connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
     try:
         queue_name = "test-queue"
@@ -69,7 +82,7 @@ def consumer(pytestconfig):
 
     finally:
         consumer.close()
-        connection.close()
+        environment.close()
 
 
 class ConsumerTestException(BaseException):