adding match unfiltered test

Author: DanielePalaia

examples/getting_started/main.py renamed to examples/getting_started/basic-example.py

@@ -19,7 +19,12 @@ def __init__(self):
         self._count = 0
 
     def on_message(self, event: Event):
-        print("received message: " + str(event.message.body))
+        print(
+            "received message from stream: "
+            + str(event.message.body)
+            + " with offset: "
+            + str(event.message.annotations["x-stream-offset"])
+        )
 
         # accepting
         self.delivery_context.accept(event)
@@ -92,6 +97,7 @@ def main() -> None:
 
     stream_filter_options = StreamFilterOptions()
     # can be first, last, next or an offset long
+    # you can also specify stream filters
     stream_filter_options.offset(OffsetSpecification.first)
 
     consumer = consumer_connection.consumer(
@@ -118,7 +124,7 @@ def main() -> None:
 
     #
     print("delete queue")
-    # management.delete_queue(queue_name)
+    management.delete_queue(queue_name)
 
     print("closing connections")
     management.close()

examples/getting_started/example_with_streams.py

@@ -266,3 +266,40 @@ def test_stream_filtering_not_present(connection: Connection) -> None:
     management.delete_queue(stream_name)
 
     assert raised is True
+
+
+def test_stream_match_unfiltered(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"])
+        stream_filter_options.filter_match_unfiltered(True)
+        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)
+        consumer.run()
+    # ack to terminate the consumer
+    except ConsumerTestException:
+        pass
+
+    consumer.close()
+
+    management.delete_queue(stream_name)