Support x-stream-match-unfiltered in Stomp

Author: acogoluegnes

deps/rabbitmq_stomp/include/rabbit_stomp_headers.hrl

@@ -28,6 +28,7 @@
 -define(HEADER_PREFETCH_COUNT, "prefetch-count").
 -define(HEADER_X_STREAM_OFFSET, "x-stream-offset").
 -define(HEADER_X_STREAM_FILTER, "x-stream-filter").
+-define(HEADER_X_STREAM_MATCH_UNFILTERED, "x-stream-match-unfiltered").
 -define(HEADER_PRIORITY, "priority").
 -define(HEADER_RECEIPT, "receipt").
 -define(HEADER_REDELIVERED, "redelivered").

deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl

@@ -717,7 +717,9 @@ do_subscribe(Destination, DestHdr, Frame,
     end.
 
 subscribe_arguments(Frame) ->
-    subscribe_arguments([?HEADER_X_STREAM_OFFSET, ?HEADER_X_STREAM_FILTER], Frame, []).
+    subscribe_arguments([?HEADER_X_STREAM_OFFSET,
+                         ?HEADER_X_STREAM_FILTER,
+                         ?HEADER_X_STREAM_MATCH_UNFILTERED], Frame, []).
 
 subscribe_arguments([], _Frame , Acc) ->
     Acc;
@@ -731,15 +733,23 @@ subscribe_argument(?HEADER_X_STREAM_OFFSET, Frame, Acc) ->
         not_found ->
             Acc;
         {OffsetType, OffsetValue} ->
-            [{<<"x-stream-offset">>, OffsetType, OffsetValue}] ++ Acc
+            [{list_to_binary(?HEADER_X_STREAM_OFFSET), OffsetType, OffsetValue}] ++ Acc
     end;
 subscribe_argument(?HEADER_X_STREAM_FILTER, Frame, Acc) ->
     StreamFilter = rabbit_stomp_frame:stream_filter_header(Frame),
     case StreamFilter of
         not_found ->
             Acc;
         {FilterType, FilterValue} ->
-            [{<<"x-stream-filter">>, FilterType, FilterValue}] ++ Acc
+            [{list_to_binary(?HEADER_X_STREAM_FILTER), FilterType, FilterValue}] ++ Acc
+    end;
+subscribe_argument(?HEADER_X_STREAM_MATCH_UNFILTERED, Frame, Acc) ->
+    MatchUnfiltered = rabbit_stomp_frame:boolean_header(Frame, ?HEADER_X_STREAM_MATCH_UNFILTERED),
+    case MatchUnfiltered of
+        {ok, MU} ->
+            [{list_to_binary(?HEADER_X_STREAM_MATCH_UNFILTERED), bool, MU}] ++ Acc;
+        not_found ->
+            Acc
     end.
 
 check_subscription_access(Destination = {topic, _Topic},

deps/rabbitmq_stomp/test/python_SUITE_data/src/x_queue_type_stream.py

@@ -28,6 +28,7 @@ def test_stream_queue(self):
                     'x-max-age' : '10h',
                     'x-stream-max-segment-size-bytes' : 1048576,
                     'x-stream-filter-size-bytes' : 32,
+                    'x-stream-match-unfiltered' : True,
                     'durable': True,
                     'auto-delete': False,
                     'id': 1234,