Consumer implementation (#15)

* connection layer implementation

* WIP

* implementing Declare Exchange/Queue

* test for gabriele

* body encodiding

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

* adding basic tests

* implementing bind

* adding user-defined exception

* Adding debugging info

* publisher implementation

* adding publisher basic test

* improve help_address utility functions

* modify example

* integrate qpid-proton lib

* implementing purge operation

* improving url helper

* cleaning up qpid unecessary folders and files

* some improvements

* implementing queue_info

* fixing queue arguments management

* better management of arguments

* improved arguments management during declare_queue

* adding purge test

* adding fixtures in tests

* adding a publisher test

* removing useless queue_type parameter

* removing receiver from publisher

* consumer implementation

* Implementing Connection, Management and Publisher modules (#10)

* connection layer implementation

* WIP

* implementing Declare Exchange/Queue

* test for gabriele

* body encodiding

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

* adding basic tests

* implementing bind

* adding user-defined exception

* Adding debugging info

* publisher implementation

* adding publisher basic test

* improve help_address utility functions

* modify example

* integrate qpid-proton lib

* implementing purge operation

* improving url helper

* cleaning up qpid unecessary folders and files

* some improvements

* implementing queue_info

* fixing queue arguments management

* better management of arguments

* improved arguments management during declare_queue

* adding purge test

* adding fixtures in tests

* adding a publisher test

* removing useless queue_type parameter

* removing receiver from publisher

---------

Signed-off-by: Gabriele Santomaggio <[email protected]>
Co-authored-by: DanielePalaia <daniele985@@gmail.com>
Co-authored-by: Gabriele Santomaggio <[email protected]>

* consumer implementation

* modification for testing purpose

* change option

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

* adding tests and improvementes

* improving ack implementation

* adding requeue with annotations test

* adding dead lettering test

* adding discard with annotations test

* refactoring tests

* improving AddressHelper utility functions

* Change the class name for the consumer

add documentation

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

* adding validations for annotations

---------

Signed-off-by: Gabriele Santomaggio <[email protected]>
Co-authored-by: DanielePalaia <daniele985@@gmail.com>
Co-authored-by: Gabriele Santomaggio <[email protected]>

Author: 3 people

examples/getting_started/main.py

@@ -1,28 +1,89 @@
+# type: ignore
+
+
 from rabbitmq_amqp_python_client import (
+    AddressHelper,
+    AMQPMessagingHandler,
     BindingSpecification,
     Connection,
+    Event,
     ExchangeSpecification,
     Message,
     QuorumQueueSpecification,
-    exchange_address,
 )
 
 
+class MyMessageHandler(AMQPMessagingHandler):
+
+    def __init__(self):
+        super().__init__()
+        self._count = 0
+
+    def on_message(self, event: Event):
+        print("received message: " + str(event.message.annotations))
+
+        # accepting
+        self.delivery_context.accept(event)
+
+        # in case of rejection (+eventually deadlettering)
+        # self.delivery_context.discard(event)
+
+        # in case of requeuing
+        # self.delivery_context.requeue(event)
+
+        # annotations = {}
+        # annotations[symbol('x-opt-string')] = 'x-test1'
+        # in case of requeuing with annotations added
+        # self.delivery_context.requeue_with_annotations(event, annotations)
+
+        # in case of rejection with annotations added
+        # self.delivery_context.discard_with_annotations(event)
+
+        print("count " + str(self._count))
+
+        self._count = self._count + 1
+
+        if self._count == 100:
+            print("closing receiver")
+            # if you want you can add cleanup operations here
+            # event.receiver.close()
+            # event.connection.close()
+
+    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() -> 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 = 100
 
     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)
+        # QuorumQueueSpecification(name=queue_name, dead_letter_exchange="dead-letter")
+    )
 
     print("binding queue to exchange")
     bind_name = management.bind(
@@ -33,21 +94,44 @@ def main() -> None:
         )
     )
 
-    addr = exchange_address(exchange_name, routing_key)
+    addr = AddressHelper.exchange_address(exchange_name, routing_key)
+
+    addr_queue = AddressHelper.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))
+    # management.close()
+
+    # 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 - press control + c to terminate to consume"
+    )
+    consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
+
+    try:
+        consumer.run()
+    except KeyboardInterrupt:
+        pass
+
+    print("cleanup")
+    consumer.close()
+    # once we finish consuming if we close the connection we need to create a new one
+    # connection = create_connection()
+    # management = connection.management()
+
     print("unbind")
     management.unbind(bind_name)
 
-    print("purging the queue")
-    management.purge_queue(queue_name)
-
     print("delete queue")
     management.delete_queue(queue_name)
 
@@ -56,7 +140,9 @@ def main() -> None:
 
     print("closing connections")
     management.close()
+    print("after management closing")
     connection.close()
+    print("after connection closing")
 
 
 if __name__ == "__main__":

rabbitmq_amqp_python_client/__init__.py

@@ -1,15 +1,22 @@
 from importlib import metadata
 
-from .address_helper import exchange_address, queue_address
-from .common import QueueType
+from .address_helper import AddressHelper
+from .amqp_consumer_handler import AMQPMessagingHandler
+from .common import ExchangeType, QueueType
 from .connection import Connection
+from .consumer import Consumer
 from .entities import (
     BindingSpecification,
     ExchangeSpecification,
 )
+from .exceptions import ArgumentOutOfRangeException
 from .management import Management
 from .publisher import Publisher
+from .qpid.proton._data import symbol  # noqa: E402
+from .qpid.proton._delivery import Delivery
+from .qpid.proton._events import Event
 from .qpid.proton._message import Message
+from .qpid.proton.handlers import MessagingHandler
 from .queues import (
     ClassicQueueSpecification,
     QuorumQueueSpecification,
@@ -35,7 +42,14 @@
     "BindingSpecification",
     "QueueType",
     "Publisher",
-    "exchange_address",
-    "queue_address",
     "Message",
+    "Consumer",
+    "MessagingHandler",
+    "Event",
+    "Delivery",
+    "symbol",
+    "ExchangeType",
+    "AddressHelper",
+    "AMQPMessagingHandler",
+    "ArgumentOutOfRangeException",
 ]

rabbitmq_amqp_python_client/address_helper.py

@@ -1,7 +1,7 @@
 from .entities import BindingSpecification
 
 
-def is_unreserved(char: str) -> bool:
+def _is_unreserved(char: str) -> bool:
     # According to RFC 3986, unreserved characters are A-Z, a-z, 0-9, '-', '.', '_', and '~'
     return char.isalnum() or char in "-._~"
 
@@ -12,7 +12,7 @@ def encode_path_segment(input_string: str) -> str:
     # Iterate over each character in the input string
     for char in input_string:
         # Check if the character is an unreserved character
-        if is_unreserved(char):
+        if _is_unreserved(char):
             encoded.append(char)  # Append as is
         else:
             # Encode character to %HH format
@@ -21,49 +21,54 @@ def encode_path_segment(input_string: str) -> str:
     return "".join(encoded)
 
 
-def exchange_address(exchange_name: str, routing_key: str = "") -> str:
-    if routing_key == "":
-        path = "/exchanges/" + encode_path_segment(exchange_name)
-    else:
-        path = (
-            "/exchanges/"
-            + encode_path_segment(exchange_name)
-            + "/"
-            + encode_path_segment(routing_key)
-        )
-
-    return path
-
+class AddressHelper:
 
-def queue_address(queue_name: str) -> str:
-    path = "/queues/" + encode_path_segment(queue_name)
-
-    return path
+    @staticmethod
+    def exchange_address(exchange_name: str, routing_key: str = "") -> str:
+        if routing_key == "":
+            path = "/exchanges/" + encode_path_segment(exchange_name)
+        else:
+            path = (
+                "/exchanges/"
+                + encode_path_segment(exchange_name)
+                + "/"
+                + encode_path_segment(routing_key)
+            )
 
+        return path
 
-def purge_queue_address(queue_name: str) -> str:
-    path = "/queues/" + encode_path_segment(queue_name) + "/messages"
+    @staticmethod
+    def queue_address(queue_name: str) -> str:
+        path = "/queues/" + encode_path_segment(queue_name)
 
-    return path
+        return path
 
+    @staticmethod
+    def purge_queue_address(queue_name: str) -> str:
+        path = "/queues/" + encode_path_segment(queue_name) + "/messages"
 
-def path_address() -> str:
-    path = "/bindings"
+        return path
 
-    return path
+    @staticmethod
+    def path_address() -> str:
+        path = "/bindings"
 
+        return path
 
-def binding_path_with_exchange_queue(bind_specification: BindingSpecification) -> str:
-    binding_path_wth_exchange_queue_key = (
-        "/bindings"
-        + "/"
-        + "src="
-        + encode_path_segment(bind_specification.source_exchange)
-        + ";"
-        + "dstq="
-        + encode_path_segment(bind_specification.destination_queue)
-        + ";key="
-        + encode_path_segment(bind_specification.binding_key)
-        + ";args="
-    )
-    return binding_path_wth_exchange_queue_key
+    @staticmethod
+    def binding_path_with_exchange_queue(
+        bind_specification: BindingSpecification,
+    ) -> str:
+        binding_path_wth_exchange_queue_key = (
+            "/bindings"
+            + "/"
+            + "src="
+            + encode_path_segment(bind_specification.source_exchange)
+            + ";"
+            + "dstq="
+            + encode_path_segment(bind_specification.destination_queue)
+            + ";key="
+            + encode_path_segment(bind_specification.binding_key)
+            + ";args="
+        )
+        return binding_path_wth_exchange_queue_key

rabbitmq_amqp_python_client/amqp_consumer_handler.py

@@ -0,0 +1,22 @@
+from .delivery_context import DeliveryContext
+from .qpid.proton.handlers import MessagingHandler
+
+"""
+AMQPMessagingHandler extends the QPID MessagingHandler.
+It is an helper to set the default values needed for manually accepting and settling messages.
+self.delivery_context is an instance of DeliveryContext, which is used to accept, reject,
+requeue or requeue with annotations a message.
+It is not mandatory to use this class, but it is a good practice to use it.
+"""
+
+
+class AMQPMessagingHandler(MessagingHandler):  # type: ignore
+
+    def __init__(self, auto_accept: bool = False, auto_settle: bool = True):
+        """
+        :param auto_accept: if True, the message is automatically accepted
+        by default is false, so the user has to manually accept the message and decide with the
+        different methods of the delivery_context what to do with the message
+        """
+        super().__init__(auto_accept=auto_accept, auto_settle=auto_settle)
+        self.delivery_context: DeliveryContext = DeliveryContext()

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