adding tests and improvementes

Author: DanielePalaia

examples/getting_started/main.py

@@ -1,69 +1,78 @@
 # type: ignore
-import time
-
 from rabbitmq_amqp_python_client import (
     BindingSpecification,
     Connection,
+    Delivery,
     Event,
     ExchangeSpecification,
     Message,
     MessagingHandler,
     QuorumQueueSpecification,
     exchange_address,
     queue_address,
-    Delivery,
 )
 
 
-
 class MyMessageHandler(MessagingHandler):
 
     def __init__(self):
         super().__init__(auto_accept=False, auto_settle=False)
+        self._count = 0
 
     def on_message(self, event: Event):
         print("received message: " + event.message.body)
 
         dlv = event.delivery
-        #dlv.update(Delivery.REJECTED)
+
         dlv.update(Delivery.ACCEPTED)
-        # dlv.settle()
+        dlv.settle()
 
+        print("count " + str(self._count))
 
-        #self.reject(event.delivery)
-        #self.settle(event.delivery, Delivery.REJECTED)
+        self._count = self._count + 1
 
+        if self._count == 100000:
+            print("closing receiver")
+            event.receiver.close()
+            event.connection.close()
 
     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:
+def create_connection() -> Connection:
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+
+    return connection
+
 
+def main() -> None:
 
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
-    connection = Connection("amqp://guest:guest@localhost:5672/")
+    messages_to_publish = 100000
 
     print("connection to amqp server")
-    connection.dial()
+    connection = create_connection()
 
     management = connection.management()
 
     print("declaring exchange and queue")
     management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
 
-    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+    management.declare_queue(
+        QuorumQueueSpecification(
+            name=queue_name, dead_letter_exchange="dead-letter-test"
+        )
+    )
 
     print("binding queue to exchange")
     bind_name = management.bind(
@@ -81,34 +90,48 @@ def main() -> None:
     print("create a publisher and publish a test message")
     publisher = connection.publisher(addr)
 
-
     print("purging the queue")
-    #messages_purged = management.purge_queue(queue_name)
+    messages_purged = management.purge_queue(queue_name)
 
-    #print("messages purged: " + str(messages_purged))
+    print("messages purged: " + str(messages_purged))
+    management.close()
 
-    # for i in range(1):
-    publisher.publish(Message(body="test"))
+    # publish 10 messages
+    for i in range(messages_to_publish):
+        publisher.publish(Message(body="test"))
 
     publisher.close()
 
-    print("create a consumer and consume the test message")
-
+    print(
+        "create a consumer and consume the test message - press control + c to terminate to consume"
+    )
     consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
 
+    try:
+        consumer.run()
+    except KeyboardInterrupt:
+        pass
+
+    print("cleanup")
+    # once we finish consuming we close the connection so we need to create a new one
+    connection = create_connection()
+
+    management = connection.management()
     print("unbind")
     management.unbind(bind_name)
 
-    #consumer.close()
     print("delete queue")
-    # management.delete_queue(queue_name)
+    management.delete_queue(queue_name)
 
     print("delete exchange")
     management.delete_exchange(exchange_name)
 
     print("closing connections")
     management.close()
+    consumer.close()
+    print("after management closing")
     connection.close()
+    print("after connection closing")
 
 
 if __name__ == "__main__":

rabbitmq_amqp_python_client/__init__.py

@@ -10,6 +10,7 @@
 )
 from .management import Management
 from .publisher import Publisher
+from .qpid.proton._delivery import Delivery
 from .qpid.proton._events import Event
 from .qpid.proton._message import Message
 from .qpid.proton.handlers import MessagingHandler
@@ -19,8 +20,6 @@
     StreamSpecification,
 )
 
-from .qpid.proton._delivery import Delivery
-
 try:
     __version__ = metadata.version(__package__)
     __license__ = metadata.metadata(__package__)["license"]

rabbitmq_amqp_python_client/consumer.py

@@ -1,7 +1,7 @@
 import logging
 from typing import Optional
 
-from .options import ReceiverOption
+from .options import ReceiverOptionUnsettled
 from .qpid.proton._handlers import MessagingHandler
 from .qpid.proton._message import Message
 from .qpid.proton.utils import (
@@ -35,13 +35,20 @@ def consume(self) -> Message:
             return self._receiver.receive()
 
     def close(self) -> None:
-        logger.debug("Closing Sender and Receiver")
+        logger.debug("Closing the receiver")
         if self._receiver is not None:
             self._receiver.close()
+
+    def run(self) -> None:
         if self._receiver is not None:
-            self._receiver.close()
+            self._receiver.container.run()
+
+    def stop(self) -> None:
+        if self._receiver is not None:
+            self._receiver.container.stop_events()
+            self._receiver.container.stop()
 
     def _create_receiver(self, addr: str) -> BlockingReceiver:
         return self._conn.create_receiver(
-            addr, options=ReceiverOption(addr), handler=self._handler
+            addr, options=ReceiverOptionUnsettled(addr), handler=self._handler
         )

rabbitmq_amqp_python_client/consumer.pytmp

@@ -25,6 +25,18 @@ class ReceiverOption(LinkOption):  # type: ignore
     def __init__(self, addr: str):
         self._addr = addr
 
+    def apply(self, link: Link) -> None:
+        link.target.address = self._addr
+        link.snd_settle_mode = Link.SND_SETTLED
+        link.rcv_settle_mode = Link.RCV_FIRST
+        link.properties = PropertyDict({symbol("paired"): True})
+        link.source.dynamic = False
+
+
+class ReceiverOptionUnsettled(LinkOption):  # type: ignore
+    def __init__(self, addr: str):
+        self._addr = addr
+
     def apply(self, link: Link) -> None:
         link.target.address = self._addr
         link.snd_settle_mode = Link.SND_UNSETTLED

rabbitmq_amqp_python_client/options.py

@@ -247,7 +247,6 @@ class IncomingMessageHandler(Handler, Acking):
     def __init__(
         self, auto_accept: bool = True, delegate: Optional[Handler] = None
     ) -> None:
-        print("auto accept is" + str(auto_accept))
         self.delegate = delegate
         self.auto_accept = auto_accept
 
@@ -266,10 +265,8 @@ def on_delivery(self, event: Event) -> None:
                     dlv.settle()
             else:
                 try:
-                    print("sending on message")
                     self.on_message(event)
                     if self.auto_accept:
-                        print("accepting")
                         dlv.update(Delivery.ACCEPTED)
                         dlv.settle()
                 except Reject:

rabbitmq_amqp_python_client/qpid/proton/_handlers.py

@@ -1,6 +1,11 @@
 import pytest
 
-from rabbitmq_amqp_python_client import Connection
+from rabbitmq_amqp_python_client import (
+    Connection,
+    Delivery,
+    Event,
+    MessagingHandler,
+)
 
 
 @pytest.fixture()
@@ -25,3 +30,87 @@ def management(pytestconfig):
     finally:
         management.close()
         connection.close()
+
+
+class ConsumerTestException(BaseException):
+    # Constructor or Initializer
+    def __init__(self, msg: str):
+        self.msg = msg
+
+    def __str__(self) -> str:
+        return repr(self.msg)
+
+
+class MyMessageHandlerAccept(MessagingHandler):
+
+    def __init__(self):
+        super().__init__(auto_accept=True, auto_settle=True)
+        self._received = 0
+
+    def on_message(self, event: Event):
+        self._received = self._received + 1
+        if self._received == 1000:
+            event.connection.close()
+            raise ConsumerTestException("consumed")
+
+
+class MyMessageHandlerNoack(MessagingHandler):
+
+    def __init__(self):
+        super().__init__(auto_accept=False, auto_settle=False)
+        self._received = 0
+
+    def on_message(self, event: Event):
+        self._received = self._received + 1
+        if self._received == 1000:
+            event.receiver.close()
+            event.connection.close()
+            raise ConsumerTestException("consumed")
+
+    def on_connection_closed(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")
+
+
+class MyMessageHandlerReject(MessagingHandler):
+
+    def __init__(self):
+        super().__init__(auto_accept=False, auto_settle=True)
+        self._received = 0
+
+    def on_message(self, event: Event):
+        dlv = event.delivery
+        dlv.update(Delivery.REJECTED)
+        dlv.settle()
+        self._received = self._received + 1
+        if self._received == 1000:
+            event.connection.close()
+            raise ConsumerTestException("consumed")
+
+
+class MyMessageHandlerRequeue(MessagingHandler):
+
+    def __init__(self):
+        super().__init__(auto_accept=False, auto_settle=True)
+        self._received = 0
+
+    def on_message(self, event: Event):
+        dlv = event.delivery
+        dlv.update(Delivery.RELEASED)
+        dlv.settle()
+        self._received = self._received + 1
+        if self._received == 1000:
+            event.connection.close()
+            raise ConsumerTestException("consumed")
+
+
+def create_connection() -> Connection:
+    connection_consumer = Connection("amqp://guest:guest@localhost:5672/")
+    connection_consumer.dial()
+
+    return connection_consumer