Extend AMQP message interceptor test

Author: ansd

deps/rabbit/test/amqp_client_SUITE.erl

@@ -2895,28 +2895,51 @@ available_messages(QType, Config) ->
 
 incoming_message_interceptors(Config) ->
     Key = {rabbit, ?FUNCTION_NAME},
-    ok = rpc(Config, persistent_term, put, [Key, [{set_header_routing_node, false}]]),
-
-    OpnConf = connection_config(Config),
-    {ok, Connection} = amqp10_client:open_connection(OpnConf),
-    {ok, Session} = amqp10_client:begin_session_sync(Connection),
-    QName = atom_to_binary(?FUNCTION_NAME),
-    Address = <<"/queue/", QName/binary>>,
-    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"test-receiver">>, Address),
-    Address = <<"/queue/", QName/binary>>,
-    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"test-sender">>, Address, settled),
+    ok = rpc(Config, persistent_term, put, [Key, [{set_header_routing_node, false},
+                                                  {set_header_timestamp, false}]]),
+    Stream = <<"my stream">>,
+    QQName = <<"my quorum queue">>,
+    {Connection, Session, LinkPair} = init(Config),
+    {ok, #{type := <<"stream">>}} = rabbitmq_amqp_client:declare_queue(
+                                      LinkPair,
+                                      Stream,
+                                      #{arguments => #{<<"x-queue-type">> => {utf8, <<"stream">>}}}),
+    {ok, #{type := <<"quorum">>}} = rabbitmq_amqp_client:declare_queue(
+                                      LinkPair,
+                                      QQName,
+                                      #{arguments => #{<<"x-queue-type">> => {utf8, <<"quorum">>}}}),
+    ok = rabbitmq_amqp_client:bind_queue(LinkPair, Stream, <<"amq.fanout">>, <<"ignored">>, #{}),
+    ok = rabbitmq_amqp_client:bind_queue(LinkPair, QQName, <<"amq.fanout">>, <<"ignored">>, #{}),
+    {ok, StreamReceiver} = amqp10_client:attach_receiver_link(Session, <<"stream receiver">>, <<"/queue/", Stream/binary>>),
+    {ok, QQReceiver} = amqp10_client:attach_receiver_link(Session, <<"qq receiver">>, <<"/queue/", QQName/binary>>),
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, <<"/exchange/amq.fanout">>),
     ok = wait_for_credit(Sender),
-    ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag">>, <<"body">>, true)),
 
-    {ok, Msg} = amqp10_client:get_msg(Receiver),
-    ?assertEqual([<<"body">>], amqp10_msg:body(Msg)),
+    NowMillis = os:system_time(millisecond),
+    Tag = <<"tag">>,
+    ok = amqp10_client:send_msg(Sender, amqp10_msg:new(Tag, <<"body">>)),
+    ok = wait_for_settlement(Tag),
+
+    {ok, Msg1} = amqp10_client:get_msg(StreamReceiver),
+    {ok, Msg2} = amqp10_client:get_msg(QQReceiver),
+    ?assertEqual([<<"body">>], amqp10_msg:body(Msg1)),
+    ?assertEqual([<<"body">>], amqp10_msg:body(Msg2)),
+
     Node = atom_to_binary(get_node_config(Config, 0, nodename)),
-    ?assertMatch(#{<<"x-routed-by">> := Node},
-                 amqp10_msg:message_annotations(Msg)),
+    #{<<"x-routed-by">> := Node,
+      <<"x-opt-rabbitmq-received-time">> := Millis} = amqp10_msg:message_annotations(Msg1),
+    ?assertMatch(
+       #{<<"x-routed-by">> := Node,
+         <<"x-opt-rabbitmq-received-time">> := Millis},  amqp10_msg:message_annotations(Msg2)),
+    ?assert(Millis < NowMillis + 4000),
+    ?assert(Millis > NowMillis - 4000),
 
     ok = amqp10_client:detach_link(Sender),
-    ok = amqp10_client:detach_link(Receiver),
-    ok = delete_queue(Session, QName),
+    ok = amqp10_client:detach_link(StreamReceiver),
+    ok = amqp10_client:detach_link(QQReceiver),
+    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, Stream),
+    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QQName),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
     ok = end_session_sync(Session),
     ok = amqp10_client:close_connection(Connection),
     true = rpc(Config, persistent_term, erase, [Key]).