direct reply to

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

Author: Gsantomaggio

examples/direct_reply_queue/direct_reply_to.py

@@ -0,0 +1,88 @@
+# type: ignore
+
+
+from rabbitmq_amqp_python_client import (  # PosixSSlConfigurationContext,; PosixClientCert,
+    AMQPMessagingHandler,
+    Connection,
+    Converter,
+    DirectReplyToConsumerOptions,
+    Environment,
+    Event,
+    Message,
+    OutcomeState,
+)
+
+MESSAGES_TO_PUBLISH = 200
+
+
+class MyMessageHandler(AMQPMessagingHandler):
+
+    def __init__(self):
+        super().__init__()
+        self._count = 0
+
+    def on_amqp_message(self, event: Event):
+        print(
+            "received message: {} ".format(
+                Converter.bytes_to_string(event.message.body)
+            )
+        )
+
+        # accepting
+        self.delivery_context.accept(event)
+
+        self._count = self._count + 1
+        print("count " + str(self._count))
+
+        if self._count == MESSAGES_TO_PUBLISH:
+            print("received all messages")
+
+    def on_connection_closed(self, event: Event):
+        # if you want you can add cleanup operations here
+        print("connection closed")
+
+    def on_link_closed(self, event: Event) -> None:
+        # if you want you can add cleanup operations here
+        print("link closed")
+
+
+def create_connection(environment: Environment) -> Connection:
+    connection = environment.connection()
+    connection.dial()
+    return connection
+
+
+def main() -> None:
+    print("connection to amqp server")
+    environment = Environment(uri="amqp://guest:guest@localhost:5672/")
+    connection = create_connection(environment)
+    consumer = connection.consumer(message_handler=MyMessageHandler(),
+                                   consumer_options=DirectReplyToConsumerOptions())
+    addr = consumer.get_queue_address()
+    print("connecting to address: {}".format(addr))
+    publisher = create_connection(environment).publisher(addr)
+
+    for i in range(MESSAGES_TO_PUBLISH):
+        msg = Message(
+            body=Converter.string_to_bytes("test message {} ".format(i)))
+        status = publisher.publish(msg)
+        if status.remote_state == OutcomeState.ACCEPTED:
+            print("message accepted")
+        elif status.remote_state == OutcomeState.RELEASED:
+            print("message not routed")
+        elif status.remote_state == OutcomeState.REJECTED:
+            print("message not rejected")
+
+    try:
+        consumer.run()
+    except KeyboardInterrupt:
+        pass
+
+    consumer.close()
+
+    connection.close()
+    print("after connection closing")
+
+
+if __name__ == "__main__":
+    main()

rabbitmq_amqp_python_client/__init__.py

@@ -14,6 +14,7 @@
 from .consumer import Consumer
 from .entities import (
     ConsumerOptions,
+    DirectReplyToConsumerOptions,
     ExchangeCustomSpecification,
     ExchangeSpecification,
     ExchangeToExchangeBindingSpecification,
@@ -72,6 +73,7 @@
     "QuorumQueueSpecification",
     "ClassicQueueSpecification",
     "StreamSpecification",
+    "DirectReplyToConsumerOptions",
     "ExchangeToQueueBindingSpecification",
     "ExchangeToExchangeBindingSpecification",
     "QueueType",

rabbitmq_amqp_python_client/connection.py

@@ -17,6 +17,7 @@
 from .consumer import Consumer
 from .entities import (
     ConsumerOptions,
+    DirectReplyToConsumerOptions,
     OAuth2Options,
     RecoveryConfiguration,
 )
@@ -397,9 +398,13 @@ def consumer(
             Consumer: A new consumer instance
 
         Raises:
-            ArgumentOutOfRangeException: If destination address format is invalid
+            ArgumentOutOfRangeException: If destination address format is invalid.
+            Only applies if not using Direct Reply-to.
+            The server will provide the queue name in that case.
         """
-        if not validate_address(destination):
+        if not validate_address(destination) and not isinstance(
+            consumer_options, DirectReplyToConsumerOptions
+        ):
             raise ArgumentOutOfRangeException(
                 "destination address must start with /queues or /exchanges"
             )
@@ -438,9 +443,7 @@ def _on_disconnection(self) -> None:
             time.sleep(delay.total_seconds())
 
             try:
-
                 self._open_connections(reconnect_handlers=True)
-
                 self._connections.append(self)
 
             except ConnectionException as e:

rabbitmq_amqp_python_client/consumer.py

@@ -1,8 +1,14 @@
 import logging
 from typing import Literal, Optional, Union, cast
 
+from pika.exceptions import AMQPError
+
 from .amqp_consumer_handler import AMQPMessagingHandler
-from .entities import ConsumerOptions
+from .entities import (
+    ConsumerOptions,
+    DirectReplyToConsumerOptions,
+    StreamConsumerOptions,
+)
 from .options import (
     ReceiverOptionUnsettled,
     ReceiverOptionUnsettledWithFilters,
@@ -29,18 +35,17 @@ class Consumer:
         _conn (BlockingConnection): The underlying connection to RabbitMQ
         _addr (str): The address to consume from
         _handler (Optional[MessagingHandler]): Optional message handling callback
-        _stream_options (Optional[StreamConsumerOptions]): Configuration for stream consumption
+        _consumer_options (Optional[StreamConsumerOptions]): Configuration for stream consumption
         _credit (Optional[int]): Flow control credit value
     """
 
     def __init__(
-            self,
-            conn: BlockingConnection,
-            addr: str,
-            handler: Optional[AMQPMessagingHandler] = None,
-            stream_options: Optional[ConsumerOptions] = None,
-            credit: Optional[int] = None,
-            direct_reply_to: Optional[bool] = None,
+        self,
+        conn: BlockingConnection,
+        addr: str,
+        handler: Optional[AMQPMessagingHandler] = None,
+        consumer_options: Optional[ConsumerOptions] = None,
+        credit: Optional[int] = None,
     ):
         """
         Initialize a new Consumer instance.
@@ -49,27 +54,38 @@ def __init__(
             conn: The blocking connection to use for consuming
             addr: The address to consume from
             handler: Optional message handler for processing received messages
-            stream_options: Optional configuration for stream-based consumption
+            consumer_options: Optional configuration for stream-based consumption
             credit: Optional credit value for flow control
         """
         self._receiver: Optional[BlockingReceiver] = None
         self._conn = conn
         self._addr = addr
         self._handler = handler
-        self._stream_options = stream_options
+        self._consumer_options = consumer_options
         self._credit = credit
         self._consumers: list[Consumer] = []
-        self._direct_reply_to = direct_reply_to
         self._open()
 
     def _open(self) -> None:
         if self._receiver is 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 AMQPError("Receiver is not initialized")
+
     def _update_connection(self, conn: BlockingConnection) -> None:
         self._conn = conn
-        if self._stream_options is None:
+        if self._consumer_options is None:
             logger.debug("creating new receiver without stream")
             self._receiver = self._conn.create_receiver(
                 self._addr,
@@ -78,11 +94,11 @@ def _update_connection(self, conn: BlockingConnection) -> None:
             )
         else:
             logger.debug("creating new stream receiver")
-            self._stream_options.offset(self._handler.offset - 1)  # type: ignore
+            self._consumer_options.offset(self._handler.offset - 1)  # type: ignore
             self._receiver = self._conn.create_receiver(
                 self._addr,
                 options=ReceiverOptionUnsettledWithFilters(
-                    self._addr, self._stream_options
+                    self._addr, self._consumer_options
                 ),
                 handler=self._handler,
             )
@@ -145,34 +161,44 @@ def stop(self) -> None:
             self._receiver.container.stop()
 
     def _create_receiver(self, addr: str) -> BlockingReceiver:
-        logger.debug("Creating the receiver")
-        if self._direct_reply_to is None:
-            self._direct_reply_to = True
+        credit = 100
+        if self._credit is not None:
+            credit = self._credit
 
-        if self._direct_reply_to:
-            x = self._conn.create_dynamic_receiver()
-            # print(x.link.remote_source.address)
-            return x
+        if self._consumer_options is not None:
+            logger.debug(
+                "Creating the receiver, with options: %s",
+                type(self._consumer_options).__name__,
+            )
+        else:
+            logger.debug("Creating the receiver, without options")
 
-        if self._stream_options is None:
-            receiver = self._conn.create_receiver(
+        if self._consumer_options is None:
+            return self._conn.create_receiver(
                 addr,
                 options=ReceiverOptionUnsettled(addr),
                 handler=self._handler,
-                dynamic=self._direct_reply_to,
+                credit=credit,
             )
-        else:
-            receiver = self._conn.create_receiver(
+
+        if isinstance(self._consumer_options, DirectReplyToConsumerOptions):
+            print("Creating dynamic receiver for direct reply-to")
+            x = self._conn.create_dynamic_receiver(100, handler=self._handler)
+            x.credit = credit
+            return x
+
+        if isinstance(self._consumer_options, StreamConsumerOptions):
+            return self._conn.create_receiver(
                 addr,
-                options=ReceiverOptionUnsettledWithFilters(addr, self._stream_options),
+                options=ReceiverOptionUnsettledWithFilters(
+                    addr, self._consumer_options
+                ),
                 handler=self._handler,
-                dynamic=self._direct_reply_to,
             )
 
-        if self._credit is not None:
-            receiver.credit = self._credit
-
-        return receiver
+        raise AMQPError(
+            "Receiver is not initialized. No valid consumer options provided."
+        )
 
     @property
     def address(self) -> str:

rabbitmq_amqp_python_client/entities.py

@@ -160,6 +160,9 @@ def validate(self, versions: Dict[str, bool]) -> None:
     def filter_set(self) -> Dict[symbol, Described]:
         raise NotImplementedError("Subclasses should implement this method")
 
+    def direct_reply_to(self) -> bool:
+        return False
+
 
 @dataclass
 class MessageProperties:
@@ -400,6 +403,21 @@ def validate(self, versions: Dict[str, bool]) -> None:
             )
 
 
+class DirectReplyToConsumerOptions(ConsumerOptions):
+
+    def validate(self, versions: Dict[str, bool]) -> None:
+        if not versions.get("4.2.0", False):
+            raise ValidationCodeException(
+                "Direct Reply-To requires RabbitMQ 4.2.0 or higher"
+            )
+
+    def filter_set(self) -> Dict[symbol, Described]:
+        return {}
+
+    def direct_reply_to(self) -> bool:
+        return True
+
+
 @dataclass
 class RecoveryConfiguration:
     """

rabbitmq_amqp_python_client/options.py

@@ -1,9 +1,13 @@
 from .entities import ConsumerOptions
 from .qpid.proton._data import (  # noqa: E402
+    Data,
     PropertyDict,
-    symbol, Data,
+    symbol,
+)
+from .qpid.proton._endpoints import (  # noqa: E402
+    Link,
+    Terminus,
 )
-from .qpid.proton._endpoints import Link, Terminus  # noqa: E402
 from .qpid.proton.reactor import (  # noqa: E402
     Filter,
     LinkOption,
@@ -53,12 +57,10 @@ def apply(self, link: Link) -> None:
 
 
 class DynamicReceiverOption(LinkOption):  # type: ignore
-    def __init__(self):
-        pass
 
     def apply(self, link: Link) -> None:
         link.snd_settle_mode = Link.SND_SETTLED
-        link.rcv_settle_mode = Link.RCV_FIRST
+        # link.rcv_settle_mode = Link.RCV_FIRST
         link.source.expiry_policy = Terminus.EXPIRE_WITH_LINK
         link.properties = PropertyDict({symbol("paired"): True})
         link.source.dynamic = True

rabbitmq_amqp_python_client/publisher.py

@@ -94,7 +94,7 @@ def publish(self, message: Message) -> Delivery:
                 return self._sender.send(message)
         else:
             if message.address != "":
-                if validate_address(message.address) is False:
+                if not validate_address(message.address):
                     raise ArgumentOutOfRangeException(
                         "destination address must start with /queues or /exchanges"
                     )

rabbitmq_amqp_python_client/qpid/proton/_utils.py

@@ -506,9 +506,15 @@ def create_sender(
             ),
         )
 
-    def create_dynamic_receiver(self, credit: Optional[int] = None):
+    def create_dynamic_receiver(
+        self, credit: Optional[int] = None, handler: Optional[Handler] = None
+    ):
         return self.create_receiver(
-            credit=credit, dynamic=True, options=DynamicReceiverOption(), name="dynamic-receiver"
+            credit=credit,
+            dynamic=True,
+            options=DynamicReceiverOption(),
+            handler=handler,
+            name="dynamic-receiver_" + str(id(self)),
         )
 
     def create_receiver(

tests/direct_reply_to/__init__.py

@@ -0,0 +1,9 @@
+from rabbitmq_amqp_python_client import (
+    Connection,
+    DirectReplyToConsumerOptions,
+)
+
+
+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()