stream offset implementation

Author: DanielePalaia

examples/getting_started/main.py

@@ -10,6 +10,8 @@
     ExchangeSpecification,
     Message,
     QuorumQueueSpecification,
+    StreamSpecification,
+    StreamFilterOptions
 )
 
 
@@ -81,59 +83,63 @@ def main() -> None:
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
-    messages_to_publish = 100000
+    messages_to_publish = 1000
 
     print("connection to amqp server")
     connection = create_connection()
 
     management = connection.management()
 
     print("declaring exchange and queue")
-    management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
+    #management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
 
     management.declare_queue(
-        QuorumQueueSpecification(name=queue_name)
+        StreamSpecification(name=queue_name)
         # QuorumQueueSpecification(name=queue_name, dead_letter_exchange="dead-letter")
     )
 
     print("binding queue to exchange")
-    bind_name = management.bind(
-        BindingSpecification(
-            source_exchange=exchange_name,
-            destination_queue=queue_name,
-            binding_key=routing_key,
-        )
-    )
+    #bind_name = management.bind(
+    #    BindingSpecification(
+    #        source_exchange=exchange_name,
+    #        destination_queue=queue_name,
+    #        binding_key=routing_key,
+    #    )
+    #)
 
-    addr = AddressHelper.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 = connection.publisher(addr_queue)
 
     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()
 
     # publish 10 messages
     for i in range(messages_to_publish):
         status = publisher.publish(Message(body="test"))
-        if status.ACCEPTED:
-            print("message accepted")
-        elif status.RELEASED:
-            print("message not routed")
-        elif status.REJECTED:
-            print("message not rejected")
+        #if status.ACCEPTED:
+        #    print("message accepted")
+        #elif status.RELEASED:
+        #    print("message not routed")
+        #elif status.REJECTED:
+        #    print("message not rejected")
 
     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())
+
+    stream_filter_options = StreamFilterOptions()
+    stream_filter_options.offset(0)
+
+    consumer = connection.consumer(addr_queue, handler=MyMessageHandler(), stream_filter_options=stream_filter_options)
 
     try:
         consumer.run()
@@ -147,10 +153,10 @@ def main() -> None:
     # management = connection.management()
 
     print("unbind")
-    management.unbind(bind_name)
+    #management.unbind(bind_name)
 
     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

@@ -28,6 +28,8 @@
     SslConfigurationContext,
 )
 
+from .entities import StreamFilterOptions
+
 try:
     __version__ = metadata.version(__package__)
     __license__ = metadata.metadata(__package__)["license"]
@@ -61,4 +63,5 @@
     "ClientCert",
     "Delivery",
     "ConnectionClosed",
+    "StreamFilterOptions",
 ]

rabbitmq_amqp_python_client/connection.py

@@ -11,6 +11,8 @@
 from .qpid.proton.utils import BlockingConnection
 from .ssl_configuration import SslConfigurationContext
 
+from.entities import StreamFilterOptions
+
 logger = logging.getLogger(__name__)
 
 MT = TypeVar("MT")
@@ -84,11 +86,11 @@ def publisher(self, destination: str) -> Publisher:
         return publisher
 
     def consumer(
-        self, destination: str, handler: Optional[MessagingHandler] = None
+        self, destination: str, handler: Optional[MessagingHandler] = None, stream_filter_options: Optional[StreamFilterOptions] = None
     ) -> Consumer:
         if validate_address(destination) is False:
             raise ArgumentOutOfRangeException(
                 "destination address must start with /queues or /exchanges"
             )
-        consumer = Consumer(self._conn, destination, handler)
+        consumer = Consumer(self._conn, destination, handler, stream_filter_options)
         return consumer

rabbitmq_amqp_python_client/consumer.py

@@ -1,14 +1,16 @@
 import logging
 from typing import Optional
 
-from .options import ReceiverOptionUnsettled
+from .options import ReceiverOptionUnsettled, ReceiverOptionUnsettledWithFilters
 from .qpid.proton._handlers import MessagingHandler
 from .qpid.proton._message import Message
 from .qpid.proton.utils import (
     BlockingConnection,
     BlockingReceiver,
 )
 
+from.entities import StreamFilterOptions
+
 logger = logging.getLogger(__name__)
 
 
@@ -18,11 +20,13 @@ def __init__(
         conn: BlockingConnection,
         addr: str,
         handler: Optional[MessagingHandler] = None,
+        stream_options: Optional[StreamFilterOptions] = None
     ):
         self._receiver: Optional[BlockingReceiver] = None
         self._conn = conn
         self._addr = addr
         self._handler = handler
+        self._stream_options = stream_options
         self._open()
 
     def _open(self) -> None:
@@ -52,6 +56,15 @@ def stop(self) -> None:
 
     def _create_receiver(self, addr: str) -> BlockingReceiver:
         logger.debug("Creating the receiver")
-        return self._conn.create_receiver(
+        if self._stream_options is None:
+            receiver = self._conn.create_receiver(
             addr, options=ReceiverOptionUnsettled(addr), handler=self._handler
-        )
+            )
+        else:
+            print("stream option is not None")
+            receiver = self._conn.create_receiver(
+                addr, options=ReceiverOptionUnsettledWithFilters(addr, self._stream_options), handler=self._handler
+            )
+
+        return receiver
+

rabbitmq_amqp_python_client/entities.py

@@ -1,8 +1,12 @@
 from dataclasses import dataclass
-from typing import Any, Optional
+from typing import Any, Optional, Dict
 
+from .qpid.proton._data import symbol, Described
 from .common import ExchangeType, QueueType
 
+STREAM_FILTER_SPEC = "rabbitmq:stream-filter"
+STREAM_OFFSET_SPEC = "rabbitmq:stream-offset-spec"
+STREAM_FILTER_MATCH_UNFILTERED = "rabbitmq:stream-match-unfiltered"
 
 @dataclass
 class ExchangeSpecification:
@@ -33,3 +37,16 @@ class BindingSpecification:
     source_exchange: str
     destination_queue: str
     binding_key: str
+
+
+class StreamFilterOptions:
+
+    def __init__(self):
+        self._filter_set: Dict[symbol, Described] = {}
+
+    def offset(self, offset: int):
+        #self._filter_set[symbol(STREAM_FILTER_SPEC)] = Described(symbol(STREAM_FILTER_SPEC), "first")
+        self._filter_set[symbol('rabbitmq:stream-offset-spec')] = Described(symbol('rabbitmq:stream-offset-spec'), 0)
+
+    def filters(self) -> Dict[symbol, Described]:
+        return self._filter_set

rabbitmq_amqp_python_client/options.py

@@ -3,7 +3,8 @@
     symbol,
 )
 from .qpid.proton._endpoints import Link  # noqa: E402
-from .qpid.proton.reactor import LinkOption  # noqa: E402
+from .qpid.proton.reactor import LinkOption, Filter  # noqa: E402
+from .entities import StreamFilterOptions
 
 
 class SenderOption(LinkOption):  # type: ignore
@@ -52,6 +53,23 @@ 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
+        link.rcv_settle_mode = Link.RCV_FIRST
+        link.properties = PropertyDict({symbol("paired"): True})
+        link.source.dynamic = False
+
+    def test(self, link: Link) -> bool:
+        return bool(link.is_receiver)
+
+class ReceiverOptionUnsettledWithFilters(Filter):  # type: ignore
+    def __init__(self, addr: str, filter_options: StreamFilterOptions):
+        super().__init__(filter_options.filters())
+        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/qpid/proton/_reactor.py

@@ -814,6 +814,7 @@ class Filter(ReceiverOption):
     """
 
     def __init__(self, filter_set: Dict[symbol, Described] = {}) -> None:
+        print("filterset: " + str(filter_set))
         self.filter_set = filter_set
 
     def apply(self, receiver: "Receiver") -> None:

tests/conftest.py

@@ -86,9 +86,10 @@ def __init__(self):
         self._received = 0
 
     def on_message(self, event: Event):
+        print("received message: " + str(event.message.body))
         self.delivery_context.accept(event)
         self._received = self._received + 1
-        if self._received == 1000:
+        if self._received == 10:
             event.connection.close()
             raise ConsumerTestException("consumed")
 

tests/test_publisher.py

@@ -9,6 +9,7 @@
     ExchangeSpecification,
     Message,
     QuorumQueueSpecification,
+    StreamSpecification,
 )
 
 from .http_requests import delete_all_connections
@@ -238,3 +239,28 @@ def on_disconnected():
     assert disconnected is True
     assert reconnected is True
     assert message_purged == messages_to_publish - 1
+
+
+def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
+
+    stream_name = "test_stream_info_with_validation"
+    messages_to_send = 200
+
+    queue_specification = StreamSpecification(
+        name=stream_name,
+    )
+    management = connection.management()
+    management.declare_queue(queue_specification)
+
+    print("before creating publisher")
+
+    publisher = connection.publisher("/queues/" + stream_name)
+
+    print("after creating publisher")
+
+    for i in range(messages_to_send):
+
+        publisher.publish(Message(body="test"))
+
+
+

tests/test_publisher_streams.py

@@ -0,0 +1,47 @@
+from rabbitmq_amqp_python_client import (
+    StreamSpecification,
+    QueueType,
+    Management,
+    AddressHelper,
+    StreamFilterOptions,
+    Connection,
+)
+
+from .utils import publish_messages
+from .conftest import MyMessageHandlerAccept, ConsumerTestException
+
+
+def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
+
+    stream_name = "test_stream_info_with_validation"
+    messages_to_send = 200
+
+    queue_specification = StreamSpecification(
+        name=stream_name,
+    )
+    management = connection.management()
+    management.declare_queue(queue_specification)
+
+    publish_messages(connection, messages_to_send, stream_name)
+
+    addr_queue = AddressHelper.queue_address(stream_name)
+
+    stream_filter_options = StreamFilterOptions()
+    stream_filter_options.offset(0)
+
+    consumer = connection.consumer(addr_queue,  handler=MyMessageHandlerAccept())
+
+    try:
+        print("running")
+        consumer.run()
+    # ack to terminate the consumer
+    except ConsumerTestException:
+        pass
+
+    consumer.close()
+
+    management.delete_queue(stream_name)
+
+    #assert stream_info.name == stream_name
+    #assert stream_info.queue_type == QueueType.stream
+    #assert stream_info.message_count == 0