Respect message_interceptors.incoming.set_header_timestamp

When feature flag message_containers is enabled, setting
```
message_interceptors.incoming.set_header_timestamp
```
wasn't respected anymore when a message is published via MQTT to a
stream and subsequently consumed via AMQP 0.9.1.

This commit ensures that AMQP 0.9.1 header timestamp_in_ms will be
set.

Note that we must not modify the AMQP 1.0 properties section when messages
are received via AMQP 1.0 and consumed via AMQP 1.0.
Also, message annoation keys not starting with "x-" are reserved.

Author: ansd

deps/rabbit/src/mc_amqp.erl

@@ -187,14 +187,13 @@ get_property(priority, Msg) ->
 convert_to(?MODULE, Msg, _Env) ->
     Msg;
 convert_to(TargetProto, Msg, Env) ->
-    TargetProto:convert_from(?MODULE,
-                             msg_to_sections(Msg, fun (X) -> X end),
-                             Env).
+    TargetProto:convert_from(?MODULE, msg_to_sections(Msg), Env).
 
 serialize(Sections) ->
     encode_bin(Sections).
 
-protocol_state(Msg0 = #msg{header = Header0}, Anns) ->
+protocol_state(Msg0 = #msg{header = Header0,
+                           message_annotations = MA0}, Anns) ->
     Redelivered = maps:get(redelivered, Anns, false),
     FirstAcquirer = not Redelivered,
     Header = case Header0 of
@@ -203,20 +202,25 @@ protocol_state(Msg0 = #msg{header = Header0}, Anns) ->
                  #'v1_0.header'{} ->
                      Header0#'v1_0.header'{first_acquirer = FirstAcquirer}
              end,
-    Msg = Msg0#msg{header = Header},
 
-    #{?ANN_EXCHANGE := Exchange,
-      ?ANN_ROUTING_KEYS := [RKey | _]} = Anns,
-    %% any x-* annotations get added as message annotations
-    AnnsToAdd = maps:filter(fun (Key, _) -> mc_util:is_x_header(Key) end, Anns),
-
-    MACFun = fun(MAC) ->
-                     add_message_annotations(
-                       AnnsToAdd#{<<"x-exchange">> => wrap(utf8, Exchange),
-                                  <<"x-routing-key">> => wrap(utf8, RKey)}, MAC)
-             end,
-
-    msg_to_sections(Msg, MACFun).
+    MA = maps:fold(fun(?ANN_EXCHANGE, Exchange, L) ->
+                           maps_upsert(<<"x-exchange">>, {utf8, Exchange}, L);
+                      (?ANN_ROUTING_KEYS, RKeys, L) ->
+                           RKey = hd(RKeys),
+                           maps_upsert(<<"x-routing-key">>, {utf8, RKey}, L);
+                      (<<"x-", _/binary>> = K, V, L)
+                        when V =/= undefined ->
+                           %% any x-* annotations get added as message annotations
+                           maps_upsert(K, mc_util:infer_type(V), L);
+                      (<<"timestamp_in_ms">>, V, L) ->
+                           maps_upsert(<<"x-opt-rabbitmq-received-time">>, {timestamp, V}, L);
+                      (_, _, Acc) ->
+                           Acc
+                   end, MA0, Anns),
+
+    Msg = Msg0#msg{header = Header,
+                   message_annotations = MA},
+    msg_to_sections(Msg).
 
 prepare(_For, Msg) ->
     Msg.
@@ -225,13 +229,14 @@ prepare(_For, Msg) ->
 
 msg_to_sections(#msg{header = H,
                      delivery_annotations = DAC,
-                     message_annotations = MAC0,
+                     message_annotations = MAC,
                      properties = P,
                      application_properties = APC,
                      data = Data,
-                     footer = FC}, MacFun) ->
+                     footer = FC}) ->
     Tail = case FC of
-               [] -> [];
+               [] ->
+                   [];
                _ ->
                    [#'v1_0.footer'{content = FC}]
            end,
@@ -242,34 +247,40 @@ msg_to_sections(#msg{header = H,
                  Data ++ Tail
          end,
     S1 = case APC of
-             [] -> S0;
+             [] ->
+                 S0;
              _ ->
                  [#'v1_0.application_properties'{content = APC} | S0]
          end,
     S2 = case P of
-             undefined -> S1;
+             undefined ->
+                 S1;
              _ ->
                  [P | S1]
          end,
-    S3 = case MacFun(MAC0) of
-             [] -> S2;
-             MAC ->
+    S3 = case MAC of
+             [] ->
+                 S2;
+             _ ->
                  [#'v1_0.message_annotations'{content = MAC} | S2]
          end,
     S4 = case DAC of
-             [] -> S3;
+             [] ->
+                 S3;
              _ ->
                  [#'v1_0.delivery_annotations'{content = DAC} | S3]
          end,
     case H of
-        undefined -> S4;
+        undefined ->
+            S4;
         _ ->
             [H | S4]
     end.
 
-
-
-
+maps_upsert(Key, TaggedVal, KVList) ->
+    TaggedKey = {symbol, Key},
+    Elem = {TaggedKey, TaggedVal},
+    lists:keystore(TaggedKey, 1, KVList, Elem).
 
 encode_bin(undefined) ->
     <<>>;
@@ -377,22 +388,6 @@ decode([#'v1_0.amqp_value'{} = B | Rem], #msg{} = Msg) ->
     %% an amqp value can only be a singleton
     decode(Rem, Msg#msg{data = B}).
 
-add_message_annotations(Anns, MA0) ->
-    maps:fold(fun (K, V, Acc) ->
-                      map_add(symbol, K, mc_util:infer_type(V), Acc)
-              end, MA0, Anns).
-
-map_add(_T, _Key, undefined, Acc) ->
-    Acc;
-map_add(KeyType, Key, TaggedValue, Acc0) ->
-    TaggedKey = wrap(KeyType, Key),
-    lists_upsert({TaggedKey, TaggedValue}, Acc0).
-
-wrap(_Type, undefined) ->
-    undefined;
-wrap(Type, Val) ->
-    {Type, Val}.
-
 key_find(K, [{{_, K}, {_, V}} | _]) ->
     V;
 key_find(K, [_ | Rem]) ->
@@ -478,13 +473,3 @@ essential_properties(#msg{message_annotations = MA} = Msg) ->
                       Acc
               end, Anns, MA)
     end.
-
-lists_upsert(New, L) ->
-    lists_upsert(New, L, [], L).
-
-lists_upsert({Key, _} = New, [{Key, _} | Rem], Pref, _All) ->
-    lists:reverse(Pref, [New | Rem]);
-lists_upsert(New, [Item | Rem], Pref, All) ->
-    lists_upsert(New, Rem, [Item | Pref], All);
-lists_upsert(New, [], _Pref, All) ->
-    [New | All].

deps/rabbit/src/mc_amqpl.erl

@@ -149,6 +149,8 @@ convert_from(mc_amqp, Sections, _Env) ->
     %% Add remaining x- message annotations as headers
     XHeaders = lists:filtermap(fun({{symbol, <<"x-cc">>}, V}) ->
                                        {true, to_091(<<"CC">>, V)};
+                                  ({{symbol, <<"x-opt-rabbitmq-received-time">>}, {timestamp, Ts}}) ->
+                                       {true, {<<"timestamp_in_ms">>, long, Ts}};
                                   ({{symbol, <<"x-", _/binary>> = K}, V})
                                     when ?IS_SHORTSTR_LEN(K) ->
                                        case is_internal_header(K) of

deps/rabbitmq_ct_helpers/include/rabbit_assert.hrl

@@ -6,7 +6,7 @@
                       AwaitMatchResult = Expr,
                       case (AwaitMatchResult) of
                           Guard -> AwaitMatchResult;
-                          __V -> case erlang:system_time(millisecond) of
+                          __V -> case erlang:monotonic_time(millisecond) of
                                      AwaitMatchNow when AwaitMatchNow < AwaitMatchHorizon ->
                                          timer:sleep(
                                            min(PollingInterval,
@@ -21,7 +21,7 @@
                                                         {value, __V}]})
                                  end
                       end
-              end)(erlang:system_time(millisecond) + Timeout))
+              end)(erlang:monotonic_time(millisecond) + Timeout))
         end).
 
 -define(awaitMatch(Guard, Expr, Timeout),

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -1795,28 +1795,52 @@ incoming_message_interceptors(Config) ->
     Key = {rabbit, ?FUNCTION_NAME},
     ok = rpc(Config, persistent_term, put, [Key, [{set_header_timestamp, false}]]),
     Ch = rabbit_ct_client_helpers:open_channel(Config),
-    Payload = ClientId = QName = Topic = atom_to_binary(?FUNCTION_NAME),
-    declare_queue(Ch, QName, []),
-    bind(Ch, QName, Topic),
+    Payload = ClientId = Topic = atom_to_binary(?FUNCTION_NAME),
+    CQName = <<"my classic queue">>,
+    Stream = <<"my stream">>,
+    declare_queue(Ch, CQName, [{<<"x-queue-type">>, longstr, <<"classic">>}]),
+    declare_queue(Ch, Stream, [{<<"x-queue-type">>, longstr, <<"stream">>}]),
+    bind(Ch, CQName, Topic),
+    bind(Ch, Stream, Topic),
     C = connect(ClientId, Config),
-    ok = emqtt:publish(C, Topic, Payload),
+
     NowSecs = os:system_time(second),
-    NowMs = os:system_time(millisecond),
-    eventually(
-      ?_assertMatch(
-         {#'basic.get_ok'{},
-          #amqp_msg{payload = Payload,
-                    props = #'P_basic'{
-                               timestamp = Secs,
-                               headers = [{<<"timestamp_in_ms">>, long, Ms} | _XHeaders]
-                              }}}
-           when Ms < NowMs + 4000 andalso
-                Ms > NowMs - 4000 andalso
-                Secs < NowSecs + 4 andalso
-                Secs > NowSecs - 4,
-         amqp_channel:call(Ch, #'basic.get'{queue = QName}))),
+    NowMillis = os:system_time(millisecond),
+    {ok, _} = emqtt:publish(C, Topic, Payload, qos1),
 
-    delete_queue(Ch, QName),
+    {#'basic.get_ok'{},
+     #amqp_msg{payload = Payload,
+               props = #'P_basic'{
+                          timestamp = Secs,
+                          headers = [{<<"timestamp_in_ms">>, long, Millis} | _]
+                         }}
+    } = amqp_channel:call(Ch, #'basic.get'{queue = CQName}),
+
+    ?assert(Secs < NowSecs + 4),
+    ?assert(Secs > NowSecs - 4),
+    ?assert(Millis < NowMillis + 4000),
+    ?assert(Millis > NowMillis - 4000),
+
+    #'basic.qos_ok'{}  = amqp_channel:call(Ch, #'basic.qos'{prefetch_count = 1}),
+    CTag = <<"my ctag">>,
+    #'basic.consume_ok'{} = amqp_channel:subscribe(
+                              Ch,
+                              #'basic.consume'{
+                                 queue = Stream,
+                                 consumer_tag = CTag,
+                                 arguments = [{<<"x-stream-offset">>, longstr, <<"first">>}]},
+                              self()),
+    receive {#'basic.deliver'{consumer_tag = CTag},
+             #amqp_msg{payload = Payload,
+                       props = #'P_basic'{
+                                  headers = [{<<"timestamp_in_ms">>, long, Millis} | _XHeaders]
+                                 }}} ->
+                ok
+    after 5000 -> ct:fail(missing_deliver)
+    end,
+
+    delete_queue(Ch, Stream),
+    delete_queue(Ch, CQName),
     true = rpc(Config, persistent_term, erase, [Key]),
     ok = emqtt:disconnect(C).