Mc: introduce new function in mc_amqp to recover mc from stream.

This includes all the routing key recovery / defaulting needed.

Author: kjnilsson

deps/rabbit/src/mc_amqp.erl

@@ -17,6 +17,8 @@
          prepare/2
         ]).
 
+-export([init_from_stream/2]).
+
 -import(rabbit_misc,
         [maps_put_truthy/3]).
 
@@ -99,10 +101,25 @@
 
 -export_type([state/0]).
 
+%% API
+
+-spec init_from_stream(binary(), mc:annotations()) ->
+    mc:state().
+init_from_stream(Payload, #{?ANN_EXCHANGE := _,
+                            ?ANN_ROUTING_KEYS := [_|_]} =Anns0) ->
+    Sections = amqp10_framing:decode_bin(Payload, [server_mode]),
+    Msg = msg_body_encoded(Sections, Payload, #msg_body_encoded{}),
+    Anns = maps:merge(Anns0, essential_properties(Msg, recover)),
+    mc:init(?MODULE, Msg, Anns).
+
+%% CALLBACKS
+
+init(#msg_body_encoded{} = Msg) ->
+    {Msg, #{}};
 init(Payload) ->
     Sections = amqp10_framing:decode_bin(Payload, [server_mode]),
     Msg = msg_body_encoded(Sections, Payload, #msg_body_encoded{}),
-    Anns = essential_properties(Msg),
+    Anns = essential_properties(Msg, new),
     {Msg, Anns}.
 
 convert_from(?MODULE, Sections, _Env) when is_list(Sections) ->
@@ -622,7 +639,7 @@ encode_deaths(Deaths) ->
               {map, Map}
       end, Deaths).
 
-essential_properties(#msg_body_encoded{message_annotations = MA} = Msg) ->
+essential_properties(#msg_body_encoded{} = Msg, new) ->
     Durable = get_property(durable, Msg),
     Priority = get_property(priority, Msg),
     Timestamp = get_property(timestamp, Msg),
@@ -635,6 +652,9 @@ essential_properties(#msg_body_encoded{message_annotations = MA} = Msg) ->
                maps_put_truthy(
                  ttl, Ttl,
                  Anns0))),
+    Anns;
+essential_properties(#msg_body_encoded{message_annotations = MA} = Msg, recover) ->
+    Anns = essential_properties(Msg, new),
     case MA of
         [] ->
             Anns;

deps/rabbit/src/rabbit_stream_queue.erl

@@ -1305,39 +1305,12 @@ parse_uncompressed_subbatch(
     parse_uncompressed_subbatch(Rem, Offset + 1, StartOffset, QName,
                                 Name, LocalPid, Filter, Acc).
 
-entry_to_msg(Entry, Offset, #resource{kind = queue, name = QName}, Name, LocalPid, Filter) ->
-    Mc0 = mc:init(mc_amqp, Entry, #{}),
-    %% If exchange or routing keys annotation isn't present the entry most likely came
-    %% from the rabbitmq-stream plugin so we'll choose defaults that simulate use
-    %% of the direct exchange.
-    XHeaders = mc:x_headers(Mc0),
-    Exchange = case XHeaders of
-                   #{<<"x-exchange">> := {utf8, X}} ->
-                       X;
-                   _ ->
-                       <<>>
-               end,
-    RKeys0 = case XHeaders of
-                 #{<<"x-cc">> := {list, CCs}} ->
-                     [CC || {utf8, CC} <- CCs];
-                 _ ->
-                     []
-             end,
-    RKeys1 = case XHeaders of
-                 #{<<"x-routing-key">> := {utf8, RK}} ->
-                     [RK | RKeys0];
-                 _ ->
-                     RKeys0
-             end,
-    RKeys = case RKeys1 of
-                [] ->
-                    [QName];
-                _ ->
-                    RKeys1
-            end,
-    Mc1 = mc:set_annotation(?ANN_EXCHANGE, Exchange, Mc0),
-    Mc2 = mc:set_annotation(?ANN_ROUTING_KEYS, RKeys, Mc1),
-    Mc = mc:set_annotation(<<"x-stream-offset">>, Offset, Mc2),
+entry_to_msg(Entry, Offset, #resource{kind = queue, name = QName},
+             Name, LocalPid, Filter) ->
+    Mc0 = mc_amqp:init_from_stream(Entry, #{?ANN_EXCHANGE => <<>>,
+                                            ?ANN_ROUTING_KEYS => [QName],
+                                            <<"x-stream-offset">> => Offset}),
+    Mc = mc:set_annotation(<<"x-stream-offset">>, Offset, Mc0),
     case rabbit_amqp_filtex:filter(Filter, Mc) of
         true ->
             {Name, LocalPid, Offset, false, Mc};

deps/rabbit/test/mc_unit_SUITE.erl

@@ -460,11 +460,11 @@ amqpl_cc_amqp_bin_amqpl(_Config) ->
                                          {utf8, <<"q2">>}]}},
                  mc:x_headers(Msg10)),
 
-    %% routing keys and exchange annotations should be restored
-    ?assertEqual(RoutingKeys, mc:routing_keys(Msg10)),
-    ?assertEqual(<<"exch">>, mc:exchange(Msg10)),
+    %% Here, we simulate what rabbit_stream_queue does:
+    Msg10b = mc:set_annotation(?ANN_EXCHANGE, <<"exch">>, Msg10),
+    Msg10c = mc:set_annotation(?ANN_ROUTING_KEYS, [<<"apple">>, <<"q1">>, <<"q2">>], Msg10b),
 
-    MsgL2 = mc:convert(mc_amqpl, Msg10),
+    MsgL2 = mc:convert(mc_amqpl, Msg10c),
     ?assertEqual(RoutingKeys, mc:routing_keys(MsgL2)),
     ?assertMatch(#content{properties = #'P_basic'{headers = Headers}},
                  mc:protocol_state(MsgL2)).