Support stream filtering in AMQP 1.0

deps/amqp10_client/src/amqp10_client_session.erl

@@ -708,7 +708,7 @@ filter_value_type(V)
   when is_integer(V) andalso V >= 0 ->
     {uint, V};
 filter_value_type(VList) when is_list(VList) ->
-    [filter_value_type(V) || V <- VList];
+    {list, [filter_value_type(V) || V <- VList]};
 filter_value_type({T, _} = V) when is_atom(T) ->
     %% looks like an already tagged type, just pass it through
     V.
@@ -1215,7 +1215,8 @@ translate_filters_legacy_amqp_no_local_filter_test() ->
     {map,
         [{
             {symbol, <<"apache.org:no-local-filter:list">>},
-            {described, {symbol, <<"apache.org:no-local-filter:list">>}, [{utf8, <<"foo">>}, {utf8, <<"bar">>}]}
+            {described, {symbol, <<"apache.org:no-local-filter:list">>},
+             {list, [{utf8, <<"foo">>}, {utf8, <<"bar">>}]}}
         }]
     } = translate_filters(#{<<"apache.org:no-local-filter:list">> => [<<"foo">>, <<"bar">>]}).
 

deps/rabbit/src/rabbit_stream_queue.erl

@@ -549,20 +549,21 @@ deliver0(MsgId, Msg,
                          correlation = Correlation,
                          slow = Slow}, Actions}.
 
-stream_message(Msg, _FilteringSupported = true) ->
-    MsgData = msg_to_iodata(Msg),
-    case mc:x_header(<<"x-stream-filter-value">>, Msg) of
-        undefined ->
-            MsgData;
-        {utf8, Value} ->
-            {Value, MsgData}
-    end;
-stream_message(Msg, _FilteringSupported = false) ->
-    msg_to_iodata(Msg).
-
-msg_to_iodata(Msg0) ->
-    Sections = mc:protocol_state(mc:convert(mc_amqp, Msg0)),
-    mc_amqp:serialize(Sections).
+stream_message(Msg, FilteringSupported) ->
+    McAmqp = mc:convert(mc_amqp, Msg),
+    Sections = mc:protocol_state(McAmqp),
+    MsgData = mc_amqp:serialize(Sections),
+    case FilteringSupported of
+        true ->
+            case mc:x_header(<<"x-stream-filter-value">>, McAmqp) of
+                undefined ->
+                    MsgData;
+                {utf8, Value} ->
+                    {Value, MsgData}
+            end;
+        false ->
+            MsgData
+    end.
 
 -spec dequeue(_, _, _, _, client()) -> no_return().
 dequeue(_, _, _, _, #stream_client{name = Name}) ->

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -1842,20 +1842,58 @@ encode_frames(T, Msg, MaxContentLen, Transfers) ->
     end.
 
 source_filters_to_consumer_args(#'v1_0.source'{filter = {map, KVList}}) ->
-    Key = {symbol, <<"rabbitmq:stream-offset-spec">>},
-    case keyfind_unpack_described(Key, KVList) of
-        {_, {timestamp, Ts}} ->
-            [{<<"x-stream-offset">>, timestamp, Ts div 1000}]; %% 0.9.1 uses second based timestamps
-        {_, {utf8, Spec}} ->
-            [{<<"x-stream-offset">>, longstr, Spec}]; %% next, last, first and "10m" etc
-        {_, {_, Offset}} when is_integer(Offset) ->
-            [{<<"x-stream-offset">>, long, Offset}]; %% integer offset
-        _ ->
-            []
-    end;
+    source_filters_to_consumer_args(
+      [<<"rabbitmq:stream-offset-spec">>,
+       <<"rabbitmq:stream-filter">>,
+       <<"rabbitmq:stream-match-unfiltered">>],
+      KVList,
+      []);
 source_filters_to_consumer_args(_Source) ->
     [].
 
+source_filters_to_consumer_args([], _KVList, Acc) ->
+    Acc;
+source_filters_to_consumer_args([<<"rabbitmq:stream-offset-spec">> = H | T], KVList, Acc) ->
+    Key = {symbol, H},
+    Arg = case keyfind_unpack_described(Key, KVList) of
+              {_, {timestamp, Ts}} ->
+                  [{<<"x-stream-offset">>, timestamp, Ts div 1000}]; %% 0.9.1 uses second based timestamps
+              {_, {utf8, Spec}} ->
+                  [{<<"x-stream-offset">>, longstr, Spec}]; %% next, last, first and "10m" etc
+              {_, {_, Offset}} when is_integer(Offset) ->
+                  [{<<"x-stream-offset">>, long, Offset}]; %% integer offset
+              _ ->
+                  []
+          end,
+    source_filters_to_consumer_args(T, KVList, Arg ++ Acc);
+source_filters_to_consumer_args([<<"rabbitmq:stream-filter">> = H | T], KVList, Acc) ->
+    Key = {symbol, H},
+    Arg = case keyfind_unpack_described(Key, KVList) of
+              {_, {list, Filters0}} when is_list(Filters0) ->
+                  Filters = lists:foldl(fun({utf8, Filter}, L) ->
+                                                [{longstr, Filter} | L];
+                                           (_, L) ->
+                                                L
+                                        end, [], Filters0),
+                  [{<<"x-stream-filter">>, array, Filters}];
+              {_, {utf8, Filter}} ->
+                  [{<<"x-stream-filter">>, longstr, Filter}];
+              _ ->
+                  []
+          end,
+    source_filters_to_consumer_args(T, KVList, Arg ++ Acc);
+source_filters_to_consumer_args([<<"rabbitmq:stream-match-unfiltered">> = H | T], KVList, Acc) ->
+    Key = {symbol, H},
+    Arg = case keyfind_unpack_described(Key, KVList) of
+              {_, {boolean, MU}} ->
+                  [{<<"x-stream-match-unfiltered">>, bool, MU}];
+              _ ->
+                  []
+          end,
+    source_filters_to_consumer_args(T, KVList, Arg ++ Acc);
+source_filters_to_consumer_args([_ | T], KVList, Acc) ->
+    source_filters_to_consumer_args(T, KVList, Acc).
+
 keyfind_unpack_described(Key, KvList) ->
     %% filterset values _should_ be described values
     %% they aren't always however for historical reasons so we need this bit of

deps/rabbitmq_amqp1_0/test/amqp10_client_SUITE.erl

@@ -81,7 +81,8 @@ groups() ->
        resource_alarm_after_session_begin,
        max_message_size_client_to_server,
        max_message_size_server_to_client,
-       receive_transfer_flow_order
+       receive_transfer_flow_order,
+       stream_filtering
       ]},
 
      {cluster_size_3, [shuffle],
@@ -576,10 +577,9 @@ roundtrip_with_drain(Config, QueueType, QName)
     % create a receiver link
     TerminusDurability = none,
     Filter = consume_from_first(QueueType),
-    Properties = #{},
     {ok, Receiver} = amqp10_client:attach_receiver_link(
                        Session, <<"test-receiver">>, Address, unsettled,
-                       TerminusDurability, Filter, Properties),
+                       TerminusDurability, Filter),
 
     % grant credit and drain
     ok = amqp10_client:flow_link_credit(Receiver, 1, never, true),
@@ -2430,9 +2430,142 @@ queue_and_client_different_nodes(QueueLeaderNode, ClientNode, QueueType, Config)
     ok = rabbit_ct_client_helpers:close_channel(Ch),
     ok = amqp10_client:close_connection(Connection).
 
+
+stream_filtering(Config) ->
+    Stream = atom_to_binary(?FUNCTION_NAME),
+    Address = <<"/amq/queue/", Stream/binary>>,
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    amqp_channel:call(Ch, #'queue.declare'{
+                             queue = Stream,
+                             durable = true,
+                             arguments = [{<<"x-queue-type">>, longstr, <<"stream">>}]}),
+    ok = rabbit_ct_client_helpers:close_channel(Ch),
+
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session(Connection),
+    SenderLinkName = <<"test-sender">>,
+    {ok, Sender} = amqp10_client:attach_sender_link(Session,
+                                                    SenderLinkName,
+                                                    Address),
+    wait_for_credit(Sender),
+
+    %% 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.
+    WaveCount = 1000,
+    %% logic to publish a wave of messages with or without a filter value
+    Publish = fun(FilterValue) ->
+                      lists:foreach(fun(Seq) ->
+                                            {AppProps, Anns} =
+                                            case FilterValue of
+                                                undefined ->
+                                                    {#{}, #{}};
+                                                _ ->
+                                                    {#{<<"filter">> => FilterValue},
+                                                     #{<<"x-stream-filter-value">> => FilterValue}}
+                                            end,
+                                            FilterBin = rabbit_data_coercion:to_binary(FilterValue),
+                                            SeqBin = integer_to_binary(Seq),
+                                            DTag = <<FilterBin/binary, SeqBin/binary>>,
+                                            Msg0 = amqp10_msg:new(DTag, <<"my-body">>, false),
+                                            Msg1 = amqp10_msg:set_application_properties(AppProps, Msg0),
+                                            Msg2 = amqp10_msg:set_message_annotations(Anns, Msg1),
+                                            ok = amqp10_client:send_msg(Sender, Msg2),
+                                            ok = wait_for_settlement(DTag)
+                                    end, lists:seq(1, WaveCount))
+              end,
+
+    %% Publish messages with the "apple" filter value.
+    Publish(<<"apple">>),
+    %% Publish messages with no filter value.
+    Publish(undefined),
+    %% Publish messages with the "orange" filter value.
+    Publish(<<"orange">>),
+    ok = amqp10_client:detach_link(Sender),
+
+    % filtering on "apple"
+    TerminusDurability = none,
+    {ok, AppleReceiver} = amqp10_client:attach_receiver_link(
+                            Session,
+                            <<"test-receiver-1">>,
+                            Address,
+                            unsettled,
+                            TerminusDurability,
+                            #{<<"rabbitmq:stream-offset-spec">> => <<"first">>,
+                              <<"rabbitmq:stream-filter">> => <<"apple">>}),
+    ok = amqp10_client:flow_link_credit(AppleReceiver, 100, 10),
+    AppleMessages = receive_all_messages(AppleReceiver, []),
+    %% we should get less than all the waves combined
+    ?assert(length(AppleMessages) < WaveCount * 3),
+    %% client-side filtering
+    AppleFilteredMessages = lists:filter(fun(Msg) ->
+                                                 AP = amqp10_msg:application_properties(Msg),
+                                                 maps:get(<<"filter">>, AP) =:= <<"apple">>
+                                         end, AppleMessages),
+    ?assertEqual(WaveCount, length(AppleFilteredMessages)),
+    ok = amqp10_client:detach_link(AppleReceiver),
+
+    %% filtering on "apple" and "orange"
+    {ok, AppleOrangeReceiver} = amqp10_client:attach_receiver_link(
+                                  Session,
+                                  <<"test-receiver-2">>,
+                                  Address,
+                                  unsettled,
+                                  TerminusDurability,
+                                  #{<<"rabbitmq:stream-offset-spec">> => <<"first">>,
+                                    <<"rabbitmq:stream-filter">> => [<<"apple">>, <<"orange">>]}),
+    ok = amqp10_client:flow_link_credit(AppleOrangeReceiver, 100, 10),
+    AppleOrangeMessages = receive_all_messages(AppleOrangeReceiver, []),
+    %% we should get less than all the waves combined
+    ?assert(length(AppleOrangeMessages) < WaveCount * 3),
+    %% client-side filtering
+    AppleOrangeFilteredMessages = lists:filter(fun(Msg) ->
+                                                       AP = amqp10_msg:application_properties(Msg),
+                                                       Filter = maps:get(<<"filter">>, AP),
+                                                       Filter =:= <<"apple">> orelse Filter =:= <<"orange">>
+                                               end, AppleOrangeMessages),
+    ?assertEqual(WaveCount * 2, length(AppleOrangeFilteredMessages)),
+    ok = amqp10_client:detach_link(AppleOrangeReceiver),
+
+    %% filtering on "apple" and messages without a filter value
+    {ok, AppleUnfilteredReceiver} = amqp10_client:attach_receiver_link(
+                                      Session,
+                                      <<"test-receiver-3">>,
+                                      Address,
+                                      unsettled,
+                                      TerminusDurability,
+                                      #{<<"rabbitmq:stream-offset-spec">> => <<"first">>,
+                                        <<"rabbitmq:stream-filter">> => <<"apple">>,
+                                        <<"rabbitmq:stream-match-unfiltered">> => {boolean, true}}),
+    ok = amqp10_client:flow_link_credit(AppleUnfilteredReceiver, 100, 10),
+
+    AppleUnfilteredMessages = receive_all_messages(AppleUnfilteredReceiver, []),
+    %% we should get less than all the waves combined
+    ?assert(length(AppleUnfilteredMessages) < WaveCount * 3),
+    %% client-side filtering
+    AppleUnfilteredFilteredMessages = lists:filter(fun(Msg) ->
+                                                           AP = amqp10_msg:application_properties(Msg),
+                                                           not maps:is_key(<<"filter">>, AP) orelse
+                                                           maps:get(<<"filter">>, AP) =:= <<"apple">>
+                                                   end, AppleUnfilteredMessages),
+    ?assertEqual(WaveCount * 2, length(AppleUnfilteredFilteredMessages)),
+    ok = amqp10_client:detach_link(AppleUnfilteredReceiver),
+
+    delete_queue(Config, Stream),
+    ok = amqp10_client:close_connection(Connection).
+
 %% internal
 %%
 
+receive_all_messages(Receiver, Acc) ->
+    receive {amqp10_msg, Receiver, Msg} ->
+                ok = amqp10_client:accept_msg(Receiver, Msg),
+                receive_all_messages(Receiver, [Msg | Acc])
+    after 500 ->
+              lists:reverse(Acc)
+    end.
+
 connection_config(Config) ->
     connection_config(0, Config).