Translate AMQP 0.9.1 CC headers to AMQP 1.0 x-cc (#9321)

* Translate AMQP 0.9.1 CC headers to AMQP 1.0 x-cc

Translate AMQP 0.9.1 CC headers to AMQP 1.0 x-cc message annotations.

We want CC headers to be kept an AMQP legacy feature and therefore
special case its conversion to AMQP 1.0.

* Translate x-cc from 1.0 message annotation to 091 CC header

for the case where you publish via 091 with CC to a stream and consume
via 091 from a stream in which case the 091 consuming client would like
to know the original CC headers.

Author: ansd

deps/rabbit/src/mc_amqp.erl

@@ -441,7 +441,17 @@ essential_properties(#msg{message_annotations = MA} = Msg) ->
             lists:foldl(
               fun ({{symbol, <<"x-routing-key">>},
                     {utf8, Key}}, Acc) ->
-                      Acc#{routing_keys => [Key]};
+                      maps:update_with(routing_keys,
+                                       fun(L) -> [Key | L] end,
+                                       [Key],
+                                       Acc);
+                  ({{symbol, <<"x-cc">>},
+                    {list, CCs0}}, Acc) ->
+                      CCs = [CC || {_T, CC} <- CCs0],
+                      maps:update_with(routing_keys,
+                                       fun(L) -> L ++ CCs end,
+                                       CCs,
+                                       Acc);
                   ({{symbol, <<"x-exchange">>},
                     {utf8, Exchange}}, Acc) ->
                       Acc#{exchange => Exchange};

deps/rabbit/src/mc_amqpl.erl

@@ -143,9 +143,17 @@ convert_from(mc_amqp, Sections) ->
     Headers0 = [to_091(K, V) || {{utf8, K}, V} <- AP,
                                 byte_size(K) =< ?AMQP_LEGACY_FIELD_NAME_MAX_LEN],
     %% Add remaining message annotations as headers?
-    XHeaders = [to_091(K, V) || {{symbol, K}, V} <- MA,
-                                not is_internal_header(K),
-                                byte_size(K) =< ?AMQP_LEGACY_FIELD_NAME_MAX_LEN],
+    XHeaders = lists:filtermap(fun({{symbol, <<"x-cc">>}, V}) ->
+                                       {true, to_091(<<"CC">>, V)};
+                                  ({{symbol, K}, V})
+                                    when byte_size(K) =< ?AMQP_LEGACY_FIELD_NAME_MAX_LEN ->
+                                       case is_internal_header(K) of
+                                           false -> {true, to_091(K, V)};
+                                           true -> false
+                                       end;
+                                  (_) ->
+                                       false
+                               end, MA),
     {Headers1, MsgId091} = message_id(MsgId, <<"x-message-id">>, Headers0),
     {Headers, CorrId091} = message_id(CorrId, <<"x-correlation-id">>, Headers1),
 
@@ -327,14 +335,18 @@ convert_to(mc_amqp, #content{payload_fragments_rev = Payload} = Content) ->
     %% x- headers are stored as message annotations
     MA = case amqp10_section_header(?AMQP10_MESSAGE_ANNOTATIONS_HEADER, Headers) of
              undefined ->
-                 MAC0 = [{{symbol, K}, from_091(T, V)}
-                         || {K, T, V} <- Headers,
-                            mc_util:is_x_header(K),
-                            %% all message annotation keys need to be either a symbol or ulong
-                            %% but 0.9.1 field-table names are always strings
-                            is_binary(K)
-                        ],
-
+                 MAC0 = lists:filtermap(
+                          fun({<<"x-", _/binary>> = K, T, V}) ->
+                                  %% All message annotation keys need to be either a symbol or ulong
+                                  %% but 0.9.1 field-table names are always strings.
+                                  {true, {{symbol, K}, from_091(T, V)}};
+                             ({<<"CC">>, T = array, V}) ->
+                                  %% Special case the 0.9.1 CC header into 1.0 message annotations because
+                                  %% 1.0 application properties must not contain list or array values.
+                                  {true, {{symbol, <<"x-cc">>}, from_091(T, V)}};
+                             (_) ->
+                                  false
+                          end, Headers),
                  %% `type' doesn't have a direct equivalent so adding as
                  %% a message annotation here
                  MAC = map_add(symbol, <<"x-basic-type">>, utf8, Type, MAC0),

deps/rabbit/test/mc_SUITE.erl

@@ -17,6 +17,10 @@ all() ->
      {group, tests}
     ].
 
+groups() ->
+    [
+     {tests, [shuffle], all_tests()}
+    ].
 
 all_tests() ->
     [
@@ -26,16 +30,12 @@ all_tests() ->
      amqpl_table_x_header_array_of_tbls,
      amqpl_death_records,
      amqpl_amqp_bin_amqpl,
+     amqpl_cc_amqp_bin_amqpl,
      amqp_amqpl,
      amqp_to_amqpl_data_body,
      amqp_amqpl_amqp_bodies
     ].
 
-groups() ->
-    [
-     {tests, [], all_tests()}
-    ].
-
 %%%===================================================================
 %%% Test cases
 %%%===================================================================
@@ -325,9 +325,30 @@ amqpl_amqp_bin_amqpl(_Config) ->
     ?assertEqual(1, mc:ttl(MsgL2)),
     ?assertEqual({utf8, <<"apple">>}, mc:x_header(<<"x-stream-filter">>, MsgL2)),
     ?assertEqual(RoutingHeaders, mc:routing_headers(MsgL2, [])),
-
     ok.
 
+amqpl_cc_amqp_bin_amqpl(_Config) ->
+    Headers = [{<<"CC">>, array, [{longstr, <<"q1">>},
+                                  {longstr, <<"q2">>}]}],
+    Props = #'P_basic'{headers = Headers},
+    Content = #content{properties = Props,
+                       payload_fragments_rev = [<<"data">>]},
+    X = rabbit_misc:r(<<"/">>, exchange, <<"exch">>),
+    Msg = mc_amqpl:message(X, <<"apple">>, Content, #{}, true),
+
+    RoutingKeys =  [<<"apple">>, <<"q1">>, <<"q2">>],
+    ?assertEqual(RoutingKeys, mc:get_annotation(routing_keys, Msg)),
+
+    Msg10Pre = mc:convert(mc_amqp, Msg),
+    Sections = amqp10_framing:decode_bin(
+                 iolist_to_binary(amqp_serialize(Msg10Pre))),
+    Msg10 = mc:init(mc_amqp, Sections, #{}),
+    ?assertEqual(RoutingKeys, mc:get_annotation(routing_keys, Msg10)),
+
+    MsgL2 = mc:convert(mc_amqpl, Msg10),
+    ?assertEqual(RoutingKeys, mc:get_annotation(routing_keys, MsgL2)),
+    ?assertMatch(#content{properties = #'P_basic'{headers = Headers}},
+                 mc:protocol_state(MsgL2)).
 
 thead2(T, Value) ->
     {symbol(atom_to_binary(T)), {T, Value}}.