Mc: restore recovery of routing_keys and echange annotations

When initialising from an AMQP packet.

Author: kjnilsson

deps/rabbit/src/mc_amqp.erl

@@ -622,16 +622,41 @@ encode_deaths(Deaths) ->
               {map, Map}
       end, Deaths).
 
-essential_properties(Msg) ->
+essential_properties(#msg_body_encoded{message_annotations = MA} = Msg) ->
     Durable = get_property(durable, Msg),
     Priority = get_property(priority, Msg),
     Timestamp = get_property(timestamp, Msg),
     Ttl = get_property(ttl, Msg),
-    Anns = #{?ANN_DURABLE => Durable},
-    maps_put_truthy(
-      ?ANN_PRIORITY, Priority,
-      maps_put_truthy(
-        ?ANN_TIMESTAMP, Timestamp,
-        maps_put_truthy(
-          ttl, Ttl,
-          Anns))).
+    Anns0 = #{?ANN_DURABLE => Durable},
+    Anns = maps_put_truthy(
+             ?ANN_PRIORITY, Priority,
+             maps_put_truthy(
+               ?ANN_TIMESTAMP, Timestamp,
+               maps_put_truthy(
+                 ttl, Ttl,
+                 Anns0))),
+    case MA of
+        [] ->
+            Anns;
+        _ ->
+            lists:foldl(
+              fun ({{symbol, <<"x-routing-key">>},
+                    {utf8, Key}}, Acc) ->
+                      maps:update_with(?ANN_ROUTING_KEYS,
+                                       fun(L) -> [Key | L] end,
+                                       [Key],
+                                       Acc);
+                  ({{symbol, <<"x-cc">>},
+                    {list, CCs0}}, Acc) ->
+                      CCs = [CC || {_T, CC} <- CCs0],
+                      maps:update_with(?ANN_ROUTING_KEYS,
+                                       fun(L) -> L ++ CCs end,
+                                       CCs,
+                                       Acc);
+                  ({{symbol, <<"x-exchange">>},
+                    {utf8, Exchange}}, Acc) ->
+                      Acc#{?ANN_EXCHANGE => Exchange};
+                  (_, Acc) ->
+                      Acc
+              end, Anns, MA)
+    end.

deps/rabbit/test/mc_unit_SUITE.erl

@@ -460,11 +460,11 @@ amqpl_cc_amqp_bin_amqpl(_Config) ->
                                          {utf8, <<"q2">>}]}},
                  mc:x_headers(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),
+    %% routing keys and exchange annotations should be restored
+    ?assertEqual(RoutingKeys, mc:routing_keys(Msg10)),
+    ?assertEqual(<<"exch">>, mc:exchange(Msg10)),
 
-    MsgL2 = mc:convert(mc_amqpl, Msg10c),
+    MsgL2 = mc:convert(mc_amqpl, Msg10),
     ?assertEqual(RoutingKeys, mc:routing_keys(MsgL2)),
     ?assertMatch(#content{properties = #'P_basic'{headers = Headers}},
                  mc:protocol_state(MsgL2)).