consumer implementation

Author: DanielePalaia

examples/getting_started/main.py

@@ -1,13 +1,38 @@
 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 +60,35 @@ 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)

rabbitmq_amqp_python_client/__init__.py

@@ -3,13 +3,16 @@
 from .address_helper import exchange_address, queue_address
 from .common import QueueType
 from .connection import Connection
+from .consumer import Consumer
 from .entities import (
     BindingSpecification,
     ExchangeSpecification,
 )
 from .management import Management
 from .publisher import Publisher
+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 +41,7 @@
     "exchange_address",
     "queue_address",
     "Message",
+    "Consumer",
+    "MessagingHandler",
+    "Event",
 ]

rabbitmq_amqp_python_client/connection.py

@@ -1,7 +1,10 @@
 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 +37,9 @@ 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

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
+        )

rabbitmq_amqp_python_client/qpid/proton/_utils.py

@@ -272,6 +272,9 @@ def __init__(
             )
         if credit:
             receiver.flow(credit)
+
+        if fetcher is None:
+            print("fetcher is none")
         self.fetcher = fetcher
         self.container = connection.container
 
@@ -526,7 +529,7 @@ def create_receiver(
                 handler=handler or fetcher,
                 options=options,
             ),
-            fetcher,
+            handler or fetcher,
             credit=prefetch,
         )
 

tests/test_consumer.py

@@ -0,0 +1,88 @@
+import time
+
+from rabbitmq_amqp_python_client import (
+    Connection,
+    Event,
+    Message,
+    MessagingHandler,
+    QuorumQueueSpecification,
+    queue_address,
+)
+
+
+class MyMessageHandler(MessagingHandler):
+
+    def __init__(self):
+        super().__init__()
+        self._received = 0
+
+    def on_message(self, event: Event):
+        self.accept(event.delivery)
+        self._received = self._received + 1
+
+    def on_link_closed(self, event: Event) -> None:
+        assert self._received > 10
+
+
+def test_consumer_sync_queue(connection: Connection) -> None:
+
+    queue_name = "test-queue"
+    messages_to_send = 100
+    management = connection.management()
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    addr_queue = queue_address(queue_name)
+
+    publisher = connection.publisher("/queues/" + queue_name)
+    consumer = connection.consumer(addr_queue)
+
+    consumed = 0
+
+    # publish 10 messages
+    for i in range(messages_to_send):
+        publisher.publish(Message(body="test" + str(i)))
+
+    time.sleep(5)
+
+    # consumer synchronously without handler
+    for i in range(messages_to_send):
+        message = consumer.consume()
+        if message.body == "test" + str(i):
+            consumed = consumed + 1
+
+    assert consumed >= 10
+
+    publisher.close()
+    consumer.close()
+
+    management.delete_queue(queue_name)
+    management.close()
+
+
+def test_consumer_async_queue(connection: Connection) -> None:
+
+    messages_to_send = 1000
+
+    queue_name = "test-queue"
+
+    management = connection.management()
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    addr_queue = queue_address(queue_name)
+
+    publisher = connection.publisher("/queues/" + queue_name)
+
+    consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
+
+    # publish 10 messages
+    for i in range(messages_to_send):
+        publisher.publish(Message(body="test" + str(i)))
+
+    time.sleep(5)
+    publisher.close()
+    consumer.close()
+
+    management.delete_queue(queue_name)
+    management.close()

tests/test_publisher.py

@@ -72,8 +72,6 @@ def test_publish_exchange(connection: Connection) -> None:
 
 
 def test_publish_purge(connection: Connection) -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
 
     queue_name = "test-queue"
     management = connection.management()
@@ -84,17 +82,17 @@ def test_publish_purge(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher("/queues/" + queue_name)
-        for i in range(20):
+        for i in range(100):
             publisher.publish(Message(body="test"))
     except Exception:
         raised = True
 
-    time.sleep(4)
+    time.sleep(5)
 
     message_purged = management.purge_queue(queue_name)
 
     assert raised is False
-    assert message_purged == 20
+    assert message_purged == 100
 
     publisher.close()