Merge branch 'consumer_implementation' into consumer_imp

DanielePalaia · DanielePalaia · commit 1be8172048fb · 2025-01-14T17:24:03.000+01:00
diff --git a/examples/getting_started/main.py b/examples/getting_started/main.py
@@ -1,13 +1,40 @@
+# type: ignore
+
 from rabbitmq_amqp_python_client import (
     BindingSpecification,
     Connection,
+    Event,
     ExchangeSpecification,
     Message,
+    MessagingHandler,
     QuorumQueueSpecification,
     exchange_address,
+    queue_address,
 )
 
 
+class MyMessageHandler(MessagingHandler):
+
+    def __init__(self):
+        super().__init__()
+
+    def on_message(self, event: Event):
+        print("received message: " + event.message.body)
+        self.accept(event.delivery)
+
+    def on_connection_closed(self, event: Event):
+        print("connection closed")
+
+    def on_connection_cloing(self, event: Event):
+        print("connection closed")
+
+    def on_link_closed(self, event: Event) -> None:
+        print("link closed")
+
+    def on_rejected(self, event: Event) -> None:
+        print("rejected")
+
+
 def main() -> None:
     exchange_name = "test-exchange"
     queue_name = "example-queue"
@@ -35,21 +62,33 @@ def main() -> None:
 
     addr = exchange_address(exchange_name, routing_key)
 
+    addr_queue = queue_address(queue_name)
+
     print("create a publisher and publish a test message")
     publisher = connection.publisher(addr)
 
     publisher.publish(Message(body="test"))
 
+    print("purging the queue")
+    messages_purged = management.purge_queue(queue_name)
+
+    print("messages purged: " + str(messages_purged))
+
+    for i in range(10):
+        publisher.publish(Message(body="test"))
+
     publisher.close()
 
+    print("create a consumer and consume the test message")
+
+    consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
+
     print("unbind")
     management.unbind(bind_name)
 
-    print("purging the queue")
-    management.purge_queue(queue_name)
-
+    consumer.close()
     print("delete queue")
-    management.delete_queue(queue_name)
+    # management.delete_queue(queue_name)
 
     print("delete exchange")
     management.delete_exchange(exchange_name)
diff --git a/examples/getting_started/main.pytmp b/examples/getting_started/main.pytmp
@@ -0,0 +1,102 @@
+# type: ignore
+
+from rabbitmq_amqp_python_client import (
+    BindingSpecification,
+    Connection,
+    Event,
+    ExchangeSpecification,
+    Message,
+    MessagingHandler,
+    QuorumQueueSpecification,
+    exchange_address,
+    queue_address,
+)
+
+
+class MyMessageHandler(MessagingHandler):
+
+    def __init__(self):
+        super().__init__()
+
+    def on_message(self, event: Event):
+        print("received message: " + event.message.body)
+        self.accept(event.delivery)
+
+    def on_connection_closed(self, event: Event):
+        print("connection closed")
+
+    def on_connection_cloing(self, event: Event):
+        print("connection closed")
+
+    def on_link_closed(self, event: Event) -> None:
+        print("link closed")
+
+    def on_rejected(self, event: Event) -> None:
+        print("rejected")
+
+
+def main() -> None:
+    exchange_name = "test-exchange"
+    queue_name = "example-queue"
+    routing_key = "routing-key"
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+
+    print("connection to amqp server")
+    connection.dial()
+
+    management = connection.management()
+
+    print("declaring exchange and queue")
+    management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    print("binding queue to exchange")
+    bind_name = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+            binding_key=routing_key,
+        )
+    )
+
+    addr = exchange_address(exchange_name, routing_key)
+
+    addr_queue = queue_address(queue_name)
+
+    print("create a publisher and publish a test message")
+    publisher = connection.publisher(addr)
+
+    publisher.publish(Message(body="test"))
+
+    print("purging the queue")
+    messages_purged = management.purge_queue(queue_name)
+
+    print("messages purged: " + str(messages_purged))
+
+    for i in range(10):
+        publisher.publish(Message(body="test"))
+
+    publisher.close()
+
+    print("create a consumer and consume the test message")
+
+    consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
+
+    print("unbind")
+    management.unbind(bind_name)
+
+    consumer.close()
+    print("delete queue")
+    # management.delete_queue(queue_name)
+
+    print("delete exchange")
+    management.delete_exchange(exchange_name)
+
+    print("closing connections")
+    management.close()
+    connection.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/rabbitmq_amqp_python_client/__init__.py b/rabbitmq_amqp_python_client/__init__.py
@@ -3,13 +3,22 @@
 from .address_helper import exchange_address, queue_address
 from .common import QueueType
 from .connection import Connection
+<<<<<<< HEAD
+=======
+from .consumer import Consumer
+>>>>>>> consumer_implementation
 from .entities import (
     BindingSpecification,
     ExchangeSpecification,
 )
 from .management import Management
 from .publisher import Publisher
+
+from .qpid.proton._message import Message
+
+from .qpid.proton._events import Event
 from .qpid.proton._message import Message
+from .qpid.proton.handlers import MessagingHandler
 from .queues import (
     ClassicQueueSpecification,
     QuorumQueueSpecification,
@@ -38,4 +47,8 @@
     "exchange_address",
     "queue_address",
     "Message",
+
+    "Consumer",
+    "MessagingHandler",
+    "Event",
 ]
diff --git a/rabbitmq_amqp_python_client/connection.py b/rabbitmq_amqp_python_client/connection.py
@@ -1,7 +1,11 @@
 import logging
 
+from typing import Optional
+
+from .consumer import Consumer
 from .management import Management
 from .publisher import Publisher
+from .qpid.proton._handlers import MessagingHandler
 from .qpid.proton.utils import BlockingConnection
 
 logger = logging.getLogger(__name__)
@@ -34,3 +38,10 @@ def close(self) -> None:
     def publisher(self, destination: str) -> Publisher:
         publisher = Publisher(self._conn, destination)
         return publisher
+
+
+    def consumer(
+        self, destination: str, handler: Optional[MessagingHandler] = None
+    ) -> Consumer:
+        consumer = Consumer(self._conn, destination, handler)
+        return consumer
diff --git a/rabbitmq_amqp_python_client/consumer.py b/rabbitmq_amqp_python_client/consumer.py
@@ -0,0 +1,47 @@
+import logging
+from typing import Optional
+
+from .options import ReceiverOption
+from .qpid.proton._handlers import MessagingHandler
+from .qpid.proton._message import Message
+from .qpid.proton.utils import (
+    BlockingConnection,
+    BlockingReceiver,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Consumer:
+    def __init__(
+        self,
+        conn: BlockingConnection,
+        addr: str,
+        handler: Optional[MessagingHandler] = None,
+    ):
+        self._receiver: Optional[BlockingReceiver] = None
+        self._conn = conn
+        self._addr = addr
+        self._handler = handler
+        self._open()
+
+    def _open(self) -> None:
+        if self._receiver is None:
+            logger.debug("Creating Sender")
+            self._receiver = self._create_receiver(self._addr)
+
+    def consume(self) -> Message:
+        if self._receiver is not None:
+            return self._receiver.receive()
+
+    def close(self) -> None:
+        logger.debug("Closing Sender and Receiver")
+        if self._receiver is not None:
+            self._receiver.close()
+        if self._receiver is not None:
+            self._receiver.close()
+
+    def _create_receiver(self, addr: str) -> BlockingReceiver:
+        return self._conn.create_receiver(
+            addr, options=ReceiverOption(addr), handler=self._handler
+        )
diff --git a/rabbitmq_amqp_python_client/consumer.pytmp b/rabbitmq_amqp_python_client/consumer.pytmp
@@ -0,0 +1,47 @@
+import logging
+from typing import Optional
+
+from .options import ReceiverOption
+from .qpid.proton._handlers import MessagingHandler
+from .qpid.proton._message import Message
+from .qpid.proton.utils import (
+    BlockingConnection,
+    BlockingReceiver,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Consumer:
+    def __init__(
+        self,
+        conn: BlockingConnection,
+        addr: str,
+        handler: Optional[MessagingHandler] = None,
+    ):
+        self._receiver: Optional[BlockingReceiver] = None
+        self._conn = conn
+        self._addr = addr
+        self._handler = handler
+        self._open()
+
+    def _open(self) -> None:
+        if self._receiver is None:
+            logger.debug("Creating Sender")
+            self._receiver = self._create_receiver(self._addr)
+
+    def consume(self) -> Message:
+        if self._receiver is not None:
+            return self._receiver.receive()
+
+    def close(self) -> None:
+        logger.debug("Closing Sender and Receiver")
+        if self._receiver is not None:
+            self._receiver.close()
+        if self._receiver is not None:
+            self._receiver.close()
+
+    def _create_receiver(self, addr: str) -> BlockingReceiver:
+        return self._conn.create_receiver(
+            addr, options=ReceiverOption(addr), handler=self._handler
+        )
diff --git a/rabbitmq_amqp_python_client/qpid/proton/_utils.py b/rabbitmq_amqp_python_client/qpid/proton/_utils.py
@@ -272,6 +272,7 @@ def __init__(
             )
         if credit:
             receiver.flow(credit)
+
         self.fetcher = fetcher
         self.container = connection.container
 
@@ -526,7 +527,8 @@ def create_receiver(
                 handler=handler or fetcher,
                 options=options,
             ),
-            fetcher,
+
+            handler or fetcher,
             credit=prefetch,
         )
 
diff --git a/tests/test_consumer.py b/tests/test_consumer.py
