Merge pull request #10058 from rabbitmq/stream-filtering-in-stomp

Support stream filtering in STOMP

Author: acogoluegnes

deps/rabbitmq_stomp/include/rabbit_stomp_headers.hrl

@@ -27,6 +27,8 @@
 -define(HEADER_PERSISTENT, "persistent").
 -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").
@@ -50,6 +52,7 @@
 -define(HEADER_X_MESSAGE_TTL, "x-message-ttl").
 -define(HEADER_X_QUEUE_NAME, "x-queue-name").
 -define(HEADER_X_QUEUE_TYPE, "x-queue-type").
+-define(HEADER_X_STREAM_FILTER_SIZE_BYTES, "x-stream-filter-size-bytes").
 
 -define(MESSAGE_ID_SEPARATOR, "@@").
 
@@ -67,7 +70,8 @@
                            ?HEADER_X_MAX_LENGTH_BYTES,
                            ?HEADER_X_MAX_PRIORITY,
                            ?HEADER_X_MESSAGE_TTL,
-                           ?HEADER_X_QUEUE_TYPE
+                           ?HEADER_X_QUEUE_TYPE,
+                           ?HEADER_X_STREAM_FILTER_SIZE_BYTES
                           ]).
 
 -define(HEADER_PARAMS, [

deps/rabbitmq_stomp/src/rabbit_stomp_frame.erl

@@ -15,7 +15,7 @@
          boolean_header/2, boolean_header/3,
          integer_header/2, integer_header/3,
          binary_header/2, binary_header/3]).
--export([stream_offset_header/2]).
+-export([stream_offset_header/1, stream_filter_header/1]).
 -export([serialize/1, serialize/2]).
 
 initial_state() -> none.
@@ -211,20 +211,33 @@ binary_header(F, K) ->
 
 binary_header(F, K, D) -> default_value(binary_header(F, K), D).
 
-stream_offset_header(F, D) ->
-    case binary_header(F, ?HEADER_X_STREAM_OFFSET, D) of
-        <<"first">> ->
+stream_offset_header(F) ->
+    case binary_header(F, ?HEADER_X_STREAM_OFFSET) of
+        {ok, <<"first">>} ->
             {longstr, <<"first">>};
-        <<"last">> ->
+        {ok, <<"last">>} ->
             {longstr, <<"last">>};
-        <<"next">> ->
+        {ok, <<"next">>} ->
             {longstr, <<"next">>};
-        <<"offset=", OffsetValue/binary>> ->
+        {ok, <<"offset=", OffsetValue/binary>>} ->
             {long, binary_to_integer(OffsetValue)};
-        <<"timestamp=", TimestampValue/binary>> ->
+        {ok, <<"timestamp=", TimestampValue/binary>>} ->
             {timestamp, binary_to_integer(TimestampValue)};
         _ ->
-            D
+            not_found
+    end.
+
+stream_filter_header(F) ->
+    case binary_header(F, ?HEADER_X_STREAM_FILTER) of
+        {ok, Str} ->
+            {array, lists:reverse(
+                      lists:foldl(fun(V, Acc) ->
+                                          [{longstr, V}] ++ Acc
+                                  end,
+                                  [],
+                                  binary:split(Str, <<",">>, [global])))};
+        not_found ->
+            not_found
     end.
 
 serialize(Frame) ->

deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl

@@ -676,13 +676,7 @@ do_subscribe(Destination, DestHdr, Frame,
                     {stop, normal, close_connection(State)};
                 error ->
                     ExchangeAndKey = parse_routing(Destination, DfltTopicEx),
-                    StreamOffset = rabbit_stomp_frame:stream_offset_header(Frame, undefined),
-                    Arguments = case StreamOffset of
-                                    undefined ->
-                                        [];
-                                    {Type, Value} ->
-                                        [{<<"x-stream-offset">>, Type, Value}]
-                                end,
+                    Arguments = subscribe_arguments(Frame),
                     try
                         amqp_channel:subscribe(Channel,
                                                #'basic.consume'{
@@ -722,6 +716,42 @@ do_subscribe(Destination, DestHdr, Frame,
             Err
     end.
 
+subscribe_arguments(Frame) ->
+    subscribe_arguments([?HEADER_X_STREAM_OFFSET,
+                         ?HEADER_X_STREAM_FILTER,
+                         ?HEADER_X_STREAM_MATCH_UNFILTERED], Frame, []).
+
+subscribe_arguments([], _Frame , Acc) ->
+    Acc;
+subscribe_arguments([K | T], Frame, Acc0) ->
+    Acc1 = subscribe_argument(K, Frame, Acc0),
+    subscribe_arguments(T, Frame, Acc1).
+
+subscribe_argument(?HEADER_X_STREAM_OFFSET, Frame, Acc) ->
+    StreamOffset = rabbit_stomp_frame:stream_offset_header(Frame),
+    case StreamOffset of
+        not_found ->
+            Acc;
+        {OffsetType, OffsetValue} ->
+            [{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} ->
+            [{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},
                           #proc_state{auth_login = _User,
                                       connection = Connection,

deps/rabbitmq_stomp/src/rabbit_stomp_util.erl

@@ -296,8 +296,12 @@ build_argument(?HEADER_X_STREAM_MAX_SEGMENT_SIZE_BYTES, Val) ->
     {list_to_binary(?HEADER_X_STREAM_MAX_SEGMENT_SIZE_BYTES), long,
      list_to_integer(string:strip(Val))};
 build_argument(?HEADER_X_QUEUE_TYPE, Val) ->
-  {list_to_binary(?HEADER_X_QUEUE_TYPE), longstr,
-    list_to_binary(string:strip(Val))}.
+    {list_to_binary(?HEADER_X_QUEUE_TYPE), longstr,
+     list_to_binary(string:strip(Val))};
+build_argument(?HEADER_X_STREAM_FILTER_SIZE_BYTES, Val) ->
+    {list_to_binary(?HEADER_X_STREAM_FILTER_SIZE_BYTES), long,
+     list_to_integer(string:strip(Val))}.
+
 
 build_params(EndPoint, Headers) ->
     Params = lists:foldl(fun({K, V}, Acc) ->

deps/rabbitmq_stomp/test/frame_SUITE.erl

@@ -39,7 +39,8 @@ all() ->
     header_value_with_colon,
     headers_escaping_roundtrip,
     headers_escaping_roundtrip_without_trailing_lf,
-    stream_offset_header
+    stream_offset_header,
+    stream_filter_header
     ].
 
 parse_simple_frame(_) ->
@@ -170,17 +171,35 @@ stream_offset_header(_) ->
         {{"x-stream-offset", "next"}, {longstr, <<"next">>}},
         {{"x-stream-offset", "offset=5000"}, {long, 5000}},
         {{"x-stream-offset", "timestamp=1000"}, {timestamp, 1000}},
-        {{"x-stream-offset", "foo"}, undefined},
-        {{"some-header", "some value"}, undefined}
+        {{"x-stream-offset", "foo"}, not_found},
+        {{"some-header", "some value"}, not_found}
     ],
 
     lists:foreach(fun({Header, Expected}) ->
         ?assertEqual(
             Expected,
-            rabbit_stomp_frame:stream_offset_header(#stomp_frame{headers = [Header]}, undefined)
+            rabbit_stomp_frame:stream_offset_header(#stomp_frame{headers = [Header]})
             )
     end, TestCases).
 
+stream_filter_header(_) ->
+    TestCases = [
+        {{"x-stream-filter", "banana"}, {array, [{longstr, <<"banana">>}]}},
+        {{"x-stream-filter", "banana,apple"}, {array, [{longstr, <<"banana">>},
+                                                       {longstr, <<"apple">>}]}},
+         {{"x-stream-filter", "banana,apple,orange"}, {array, [{longstr, <<"banana">>},
+                                                               {longstr, <<"apple">>},
+                                                               {longstr, <<"orange">>}]}},
+        {{"some-header", "some value"}, not_found}
+    ],
+
+    lists:foreach(fun({Header, Expected}) ->
+        ?assertEqual(
+            Expected,
+            rabbit_stomp_frame:stream_filter_header(#stomp_frame{headers = [Header]})
+        )
+    end, TestCases).
+
 test_frame_serialization(Expected, TrailingLF) ->
     {ok, Frame, _} = parse(Expected),
     {ok, Val} = rabbit_stomp_frame:header(Frame, "head\r:\ner"),

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

@@ -27,6 +27,8 @@ def test_stream_queue(self):
                     'x-queue-type': 'stream',
                     '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,

deps/rabbitmq_stomp/test/system_SUITE.erl

@@ -34,7 +34,8 @@ groups() ->
         delete_queue_subscribe,
         temp_destination_queue,
         temp_destination_in_send,
-        blank_destination_in_send
+        blank_destination_in_send,
+        stream_filtering
     ],
 
     [{version_to_group_name(V), [sequence], Tests}
@@ -102,6 +103,13 @@ init_per_testcase0(publish_unauthorized_error, Config) ->
     StompPort = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_stomp),
     {ok, ClientFoo} = rabbit_stomp_client:connect(Version, "user", "pass", StompPort),
     rabbit_ct_helpers:set_config(Config, [{client_foo, ClientFoo}]);
+init_per_testcase0(stream_filtering, Config) ->
+    case rabbit_ct_helpers:is_mixed_versions() of
+        true ->
+            {skip, "mixed version clusters are not supported for stream filtering"};
+        _ ->
+            Config
+    end;
 init_per_testcase0(_, Config) ->
     Config.
 
@@ -310,6 +318,162 @@ blank_destination_in_send(Config) ->
     "Invalid destination" = proplists:get_value("message", Hdrs),
     ok.
 
+stream_filtering(Config) ->
+    Version = ?config(version, Config),
+    Client = ?config(stomp_client, Config),
+    Stream = atom_to_list(?FUNCTION_NAME) ++ "-" ++ integer_to_list(rand:uniform(10000)),
+    %% subscription just to create the stream from STOMP
+    SubDestination = "/topic/stream-queue-test",
+    rabbit_stomp_client:send(
+      Client, "SUBSCRIBE",
+      [{"destination", SubDestination},
+       {"receipt", "foo"},
+       {"x-queue-name", Stream},
+       {"x-queue-type", "stream"},
+       {?HEADER_X_STREAM_FILTER_SIZE_BYTES, "32"},
+       {"durable", "true"},
+       {"auto-delete", "false"},
+       {"id", "1234"},
+       {"prefetch-count", "1"},
+       {"ack", "client"}]),
+    {ok, Client1, _, _} = stomp_receive(Client, "RECEIPT"),
+    rabbit_stomp_client:send(
+      Client1, "UNSUBSCRIBE", [{"destination", SubDestination},
+                               {"id", "1234"},
+                               {"receipt", "bar"}]),
+    {ok, Client2, _, _} = stomp_receive(Client1, "RECEIPT"),
+
+    %% we are going to publish several waves of messages with and without filter values.
+    %% we will then create subscriptions with various filter options
+    %% and make sure we receive only what we asked for and not all the messages.
+
+    StreamDestination = "/amq/queue/" ++ Stream,
+    %% logic to publish a wave of messages with or without a filter value
+    WaveCount = 1000,
+    Publish =
+    fun(C, FilterValue) ->
+            lists:foldl(fun(Seq, C0) ->
+                                Headers0 = [{"destination", StreamDestination},
+                                            {"receipt", integer_to_list(Seq)}],
+                                Headers = case FilterValue of
+                                              undefined ->
+                                                  Headers0;
+                                              _ ->
+                                                  [{"x-stream-filter-value", FilterValue}] ++ Headers0
+                                          end,
+                                rabbit_stomp_client:send(
+                                  C0, "SEND", Headers, ["hello"]),
+                                {ok, C1, _, _} = stomp_receive(C0, "RECEIPT"),
+                                C1
+                        end, C, lists:seq(1, WaveCount))
+    end,
+    %% publishing messages with the "apple" filter value
+    Client3 = Publish(Client2, "apple"),
+    %% publishing messages with no filter value
+    Client4 = Publish(Client3, undefined),
+    %% publishing messages with the "orange" filter value
+    Client5 = Publish(Client4, "orange"),
+
+    %% filtering on "apple"
+    rabbit_stomp_client:send(
+      Client5, "SUBSCRIBE",
+      [{"destination", StreamDestination},
+       {"id", "0"},
+       {"ack", "client"},
+       {"prefetch-count", "1"},
+       {"x-stream-filter", "apple"},
+       {"x-stream-offset", "first"}]),
+    {Client6, AppleMessages} = stomp_receive_messages(Client5, Version),
+    %% we should get less than all the waves combined
+    ?assert(length(AppleMessages) < WaveCount * 3),
+    %% client-side filtering
+    AppleFilteredMessages =
+    lists:filter(fun(H) ->
+                         proplists:get_value("x-stream-filter-value", H) =:= "apple"
+                 end, AppleMessages),
+    %% we should have only the "apple" messages
+    ?assert(length(AppleFilteredMessages) =:= WaveCount),
+    rabbit_stomp_client:send(
+      Client6, "UNSUBSCRIBE", [{"destination", StreamDestination},
+                               {"id", "0"},
+                               {"receipt", "bar"}]),
+    {ok, Client7, _, _} = stomp_receive(Client6, "RECEIPT"),
+
+    %% filtering on "apple" and "orange"
+    rabbit_stomp_client:send(
+      Client7, "SUBSCRIBE",
+      [{"destination", StreamDestination},
+       {"id", "0"},
+       {"ack", "client"},
+       {"prefetch-count", "1"},
+       {"x-stream-filter", "apple,orange"},
+       {"x-stream-offset", "first"}]),
+    {Client8, AppleOrangeMessages} = stomp_receive_messages(Client7, Version),
+    %% we should get less than all the waves combined
+    ?assert(length(AppleOrangeMessages) < WaveCount * 3),
+    %% client-side filtering
+    AppleOrangeFilteredMessages =
+    lists:filter(fun(H) ->
+                         proplists:get_value("x-stream-filter-value", H) =:= "apple" orelse
+                         proplists:get_value("x-stream-filter-value", H) =:= "orange"
+                 end, AppleOrangeMessages),
+    %% we should have only the "apple" and "orange" messages
+    ?assert(length(AppleOrangeFilteredMessages) =:= WaveCount * 2),
+    rabbit_stomp_client:send(
+      Client8, "UNSUBSCRIBE", [{"destination", StreamDestination},
+                               {"id", "0"},
+                               {"receipt", "bar"}]),
+    {ok, Client9, _, _} = stomp_receive(Client8, "RECEIPT"),
+
+    %% filtering on "apple" and messages without a filter value
+    rabbit_stomp_client:send(
+      Client9, "SUBSCRIBE",
+      [{"destination", StreamDestination},
+       {"id", "0"},
+       {"ack", "client"},
+       {"prefetch-count", "1"},
+       {"x-stream-filter", "apple"},
+       {"x-stream-match-unfiltered", "true"},
+       {"x-stream-offset", "first"}]),
+    {Client10, AppleUnfilteredMessages} = stomp_receive_messages(Client9, Version),
+    %% we should get less than all the waves combined
+    ?assert(length(AppleUnfilteredMessages) < WaveCount * 3),
+    %% client-side filtering
+    AppleUnfilteredFilteredMessages =
+    lists:filter(fun(H) ->
+                         proplists:get_value("x-stream-filter-value", H) =:= "apple" orelse
+                         proplists:get_value("x-stream-filter-value", H) =:= undefined
+                 end, AppleUnfilteredMessages),
+    %% we should have only the "apple" messages and messages without a filter value
+    ?assert(length(AppleUnfilteredFilteredMessages) =:= WaveCount * 2),
+    rabbit_stomp_client:send(
+      Client10, "UNSUBSCRIBE", [{"destination", StreamDestination},
+                               {"id", "0"},
+                               {"receipt", "bar"}]),
+    {ok, _, _, _} = stomp_receive(Client10, "RECEIPT"),
+
+    Channel = ?config(amqp_channel, Config),
+    #'queue.delete_ok'{} = amqp_channel:call(Channel,
+                                             #'queue.delete'{queue = list_to_binary(Stream)}),
+    ok.
+
+stomp_receive_messages(Client, Version) ->
+    stomp_receive_messages(Client, [], Version).
+
+stomp_receive_messages(Client, Acc, Version) ->
+    try rabbit_stomp_client:recv(Client) of
+        {#stomp_frame{command = "MESSAGE",
+                      headers = Headers}, Client1} ->
+        MsgHeader = rabbit_stomp_util:msg_header_name(Version),
+        AckValue  = proplists:get_value(MsgHeader, Headers),
+        AckHeader = rabbit_stomp_util:ack_header_name(Version),
+        rabbit_stomp_client:send(Client1, "ACK", [{AckHeader, AckValue}]),
+        stomp_receive_messages(Client1, [Headers] ++ Acc, Version)
+    catch
+      error:{badmatch, {error, timeout}} ->
+            {Client, Acc}
+    end.
+
 stomp_receive(Client, Command) ->
     {#stomp_frame{command     = Command,
                   headers     = Hdrs,