implementing and testing filterings

Author: DanielePalaia

rabbitmq_amqp_python_client/consumer.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Optional
+from typing import Literal, Optional, Union
 
 from .entities import StreamFilterOptions
 from .options import (
@@ -36,9 +36,9 @@ def _open(self) -> None:
             logger.debug("Creating Sender")
             self._receiver = self._create_receiver(self._addr)
 
-    def consume(self) -> Message:
+    def consume(self, timeout: Union[None, Literal[False], float] = False) -> Message:
         if self._receiver is not None:
-            return self._receiver.receive()
+            return self._receiver.receive(timeout=timeout)
 
     def close(self) -> None:
         logger.debug("Closing the receiver")

tests/test_streams.py

@@ -193,3 +193,76 @@ def test_stream_read_from_offset_ten(connection: Connection) -> None:
     consumer.close()
 
     management.delete_queue(stream_name)
+
+
+def test_stream_filtering(connection: Connection) -> None:
+
+    consumer = None
+    stream_name = "test_stream_info_with_filtering"
+    messages_to_send = 10
+
+    queue_specification = StreamSpecification(
+        name=stream_name,
+    )
+    management = connection.management()
+    management.declare_queue(queue_specification)
+
+    addr_queue = AddressHelper.queue_address(stream_name)
+
+    # consume and then publish
+    try:
+        stream_filter_options = StreamFilterOptions()
+        stream_filter_options.apply_filters(["banana"])
+        connection_consumer = create_connection()
+        consumer = connection_consumer.consumer(
+            addr_queue,
+            handler=MyMessageHandlerAcceptStreamOffset(),
+            stream_filter_options=stream_filter_options,
+        )
+        # send with annotations filter banana
+        publish_messages(connection, messages_to_send, stream_name, ["banana"])
+        consumer.run()
+    # ack to terminate the consumer
+    except ConsumerTestException:
+        pass
+
+    consumer.close()
+
+    management.delete_queue(stream_name)
+
+
+def test_stream_filtering_not_present(connection: Connection) -> None:
+
+    raised = False
+    stream_name = "test_stream_info_with_filtering"
+    messages_to_send = 10
+
+    queue_specification = StreamSpecification(
+        name=stream_name,
+    )
+    management = connection.management()
+    management.declare_queue(queue_specification)
+
+    addr_queue = AddressHelper.queue_address(stream_name)
+
+    # consume and then publish
+    stream_filter_options = StreamFilterOptions()
+    stream_filter_options.apply_filters(["apple"])
+    connection_consumer = create_connection()
+    consumer = connection_consumer.consumer(
+        addr_queue, stream_filter_options=stream_filter_options
+    )
+    # send with annotations filter banana
+    publish_messages(connection, messages_to_send, stream_name, ["banana"])
+
+    try:
+        consumer.consume(timeout=1)
+    except Exception:
+        # valid no message should arrive with filter banana so a timeout exception is raised
+        raised = True
+
+    consumer.close()
+
+    management.delete_queue(stream_name)
+
+    assert raised is True

tests/utils.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from rabbitmq_amqp_python_client import (
     BindingSpecification,
     Connection,
@@ -16,11 +18,21 @@ def create_connection() -> Connection:
     return connection_consumer
 
 
-def publish_messages(connection: Connection, messages_to_send: int, queue_name) -> None:
+def publish_messages(
+    connection: Connection,
+    messages_to_send: int,
+    queue_name,
+    filters: Optional[list[str]] = None,
+) -> None:
+    annotations = {}
+    if filters is not None:
+        for filter in filters:
+            annotations = {"x-stream-filter-value": filter}
+
     publisher = connection.publisher("/queues/" + queue_name)
     # publish messages_to_send messages
     for i in range(messages_to_send):
-        publisher.publish(Message(body="test" + str(i)))
+        publisher.publish(Message(body="test" + str(i), annotations=annotations))
     publisher.close()