implement SQL Filter

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

Author: Gsantomaggio

.ci/ubuntu/gha-setup.sh

@@ -7,7 +7,7 @@ set -o xtrace
 script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
-readonly rabbitmq_image=rabbitmq:4.1.0-management
+readonly rabbitmq_image=rabbitmq:4.2-rc-management
 
 
 readonly docker_name_prefix='rabbitmq-amqp-python-client'

rabbitmq_amqp_python_client/entities.py

@@ -161,7 +161,6 @@ class MessageProperties:
     Attributes:
         message_id: Uniquely identifies a message within the system (int, UUID, bytes, or str).
         user_id: Identity of the user responsible for producing the message.
-        to: Intended destination node of the message.
         subject: Summary information about the message content and purpose.
         reply_to: Address of the node to send replies to.
         correlation_id: Client-specific id for marking or identifying messages (int, UUID, bytes, or str).
@@ -176,7 +175,6 @@ class MessageProperties:
 
     message_id: Optional[Union[int, str, bytes]] = None
     user_id: Optional[bytes] = None
-    to: Optional[str] = None
     subject: Optional[str] = None
     reply_to: Optional[str] = None
     correlation_id: Optional[Union[int, str, bytes]] = None

tests/test_streams.py

@@ -536,6 +536,82 @@ def test_stream_filter_application_properties(
     management.delete_queue(stream_name)
 
 
+class MyMessageHandlerSQLFilter(AMQPMessagingHandler):
+    def __init__(
+        self,
+    ):
+        super().__init__()
+
+    def on_message(self, event: Event):
+        self.delivery_context.accept(event)
+        assert event.message.body == Converter.string_to_bytes("the_right_one_sql")
+        assert event.message.subject == "something_in_the_filter"
+        assert event.message.reply_to == "the_reply_to"
+        assert (
+            event.message.application_properties["a_in_the_filter_key"]
+            == "a_in_the_filter_value"
+        )
+
+        raise ConsumerTestException("consumed")
+
+
+def test_stream_filter_sql(connection: Connection, environment: Environment) -> None:
+    consumer = None
+    stream_name = "test_stream_filter_sql"
+    messages_to_send = 30
+
+    queue_specification = StreamSpecification(
+        name=stream_name,
+    )
+    management = connection.management()
+    management.delete_queue(stream_name)
+    management.declare_queue(queue_specification)
+
+    addr_queue = AddressHelper.queue_address(stream_name)
+    sql = (
+        "properties.subject LIKE '%in_the_filter%' AND properties.reply_to = 'the_reply_to' "
+        "AND a_in_the_filter_key = 'a_in_the_filter_value'"
+    )
+    try:
+        connection_consumer = environment.connection()
+        connection_consumer.dial()
+        consumer = connection_consumer.consumer(
+            addr_queue,
+            message_handler=MyMessageHandlerSQLFilter(),
+            stream_consumer_options=StreamConsumerOptions(
+                filter_options=StreamFilterOptions(sql=sql)
+            ),
+        )
+        publisher = connection.publisher(addr_queue)
+        # won't match
+        for i in range(messages_to_send):
+            msg = Message(
+                body=Converter.string_to_bytes("hello_{}".format(i)),
+            )
+            publisher.publish(msg)
+
+        msqMatch = Message(
+            body=Converter.string_to_bytes("the_right_one_sql"),
+            subject="something_in_the_filter",
+            reply_to="the_reply_to",
+            application_properties={"a_in_the_filter_key": "a_in_the_filter_value"},
+        )
+
+        publisher.publish(msqMatch)
+
+        publisher.close()
+
+        consumer.run()
+        # ack to terminate the consumer
+    except ConsumerTestException:
+        pass
+
+    if consumer is not None:
+        consumer.close()
+
+    management.delete_queue(stream_name)
+
+
 class MyMessageHandlerMixingDifferentFilters(AMQPMessagingHandler):
     def __init__(
         self,