Address optional

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

rabbitmq_amqp_python_client/connection.py

@@ -17,7 +17,6 @@
 from .consumer import Consumer
 from .entities import (
     ConsumerOptions,
-    DirectReplyToConsumerOptions,
     OAuth2Options,
     RecoveryConfiguration,
 )
@@ -380,7 +379,7 @@ def publisher(self, destination: str = "") -> Publisher:
 
     def consumer(
         self,
-        destination: str,
+        destination: Optional[str] = None,
         message_handler: Optional[MessagingHandler] = None,
         consumer_options: Optional[ConsumerOptions] = None,
         credit: Optional[int] = None,
@@ -389,7 +388,7 @@ def consumer(
         Create a new consumer instance.
 
         Args:
-            destination: The address to consume from
+            destination: Optional The address to consume from
             message_handler: Optional handler for processing messages
             consumer_options: Optional configuration for queue consumption. Each queue has its own consumer options.
             credit: Optional credit value for flow control
@@ -402,9 +401,7 @@ def consumer(
             Only applies if not using Direct Reply-to.
             The server will provide the queue name in that case.
         """
-        if not validate_address(destination) and not isinstance(
-            consumer_options, DirectReplyToConsumerOptions
-        ):
+        if destination is not None and not validate_address(destination):
             raise ArgumentOutOfRangeException(
                 "destination address must start with /queues or /exchanges"
             )

rabbitmq_amqp_python_client/consumer.py

@@ -1,7 +1,6 @@
 import logging
 from typing import Literal, Optional, Union, cast
 
-
 from .amqp_consumer_handler import AMQPMessagingHandler
 from .entities import (
     ConsumerOptions,
@@ -41,7 +40,7 @@ class Consumer:
     def __init__(
         self,
         conn: BlockingConnection,
-        addr: str,
+        addr: Optional[str] = None,
         handler: Optional[AMQPMessagingHandler] = None,
         consumer_options: Optional[ConsumerOptions] = None,
         credit: Optional[int] = None,
@@ -70,34 +69,26 @@ def _open(self) -> None:
             logger.debug("Creating Receiver")
             self._receiver = self._create_receiver(self._addr)
 
-    def get_queue_address(self) -> Optional[str]:
-        """
-        Get the name of the queue from the address.
-
-        Returns:
-            str: The name of the queue.
-        """
-        if self._receiver is not None:
-            return cast(Optional[str], self._receiver.link.remote_source.address)
-        else:
-            raise Exception("Receiver is not initialized")
-
     def _update_connection(self, conn: BlockingConnection) -> None:
+        addr = ""
+        if self._addr is not None:
+            addr = self._addr
+
         self._conn = conn
         if self._consumer_options is None:
             logger.debug("creating new receiver without stream")
             self._receiver = self._conn.create_receiver(
-                self._addr,
-                options=ReceiverOptionUnsettled(self._addr),
+                addr,
+                options=ReceiverOptionUnsettled(addr),
                 handler=self._handler,
             )
         else:
             logger.debug("creating new stream receiver")
             self._consumer_options.offset(self._handler.offset - 1)  # type: ignore
             self._receiver = self._conn.create_receiver(
-                self._addr,
+                addr,
                 options=ReceiverOptionUnsettledWithFilters(
-                    self._addr, self._consumer_options
+                    addr, self._consumer_options
                 ),
                 handler=self._handler,
             )
@@ -159,7 +150,7 @@ def stop(self) -> None:
             self._receiver.container.stop_events()
             self._receiver.container.stop()
 
-    def _create_receiver(self, addr: str) -> BlockingReceiver:
+    def _create_receiver(self, addr: Optional[str] = None) -> BlockingReceiver:
         credit = 100
         if self._credit is not None:
             credit = self._credit
@@ -182,7 +173,9 @@ def _create_receiver(self, addr: str) -> BlockingReceiver:
 
         if isinstance(self._consumer_options, DirectReplyToConsumerOptions):
             logger.debug("Creating dynamic receiver for direct reply-to")
-            dynamic_receiver = self._conn.create_dynamic_receiver(100, handler=self._handler)
+            dynamic_receiver = self._conn.create_dynamic_receiver(
+                100, handler=self._handler
+            )
             dynamic_receiver.credit = credit
             return dynamic_receiver
 
@@ -195,14 +188,16 @@ def _create_receiver(self, addr: str) -> BlockingReceiver:
                 handler=self._handler,
             )
 
-        raise AMQPError(
+        raise Exception(
             "Receiver is not initialized. No valid consumer options provided."
         )
 
     @property
-    def address(self) -> str:
-        """Get the current publisher address."""
-        return self._addr
+    def address(self) -> Optional[str]:
+        if self._receiver is not None:
+            return cast(Optional[str], self._receiver.link.remote_source.address)
+        else:
+            raise Exception("Receiver is not initialized")
 
     @property
     def handler(self) -> Optional[AMQPMessagingHandler]:

rabbitmq_amqp_python_client/options.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from .entities import ConsumerOptions
 from .qpid.proton._data import (  # noqa: E402
     Data,
@@ -72,7 +74,7 @@ def apply(self, link: Link) -> None:
 
 
 class ReceiverOptionUnsettled(LinkOption):  # type: ignore
-    def __init__(self, addr: str):
+    def __init__(self, addr: Optional[str]):
         self._addr = addr
 
     def apply(self, link: Link) -> None:
@@ -87,7 +89,7 @@ def test(self, link: Link) -> bool:
 
 
 class ReceiverOptionUnsettledWithFilters(Filter):  # type: ignore
-    def __init__(self, addr: str, consumer_options: ConsumerOptions):
+    def __init__(self, addr: Optional[str], consumer_options: ConsumerOptions):
         super().__init__(consumer_options.filter_set())
         self._addr = addr
 

tests/direct_reply_to/test_direct_reply.py

@@ -5,5 +5,5 @@
 
 
 def test_consumer_create_reply_name(connection: Connection) -> None:
-    consumer = connection.consumer("", consumer_options=DirectReplyToConsumerOptions())
-    assert "/queues/amq.rabbitmq.reply-to." in consumer.get_queue_address()
+    consumer = connection.consumer(consumer_options=DirectReplyToConsumerOptions())
+    assert "/queues/amq.rabbitmq.reply-to." in consumer.address