Support responder checking for existence of requester

ansd · ansd · commit ecf3fa5e2e6a · 2025-09-03T15:23:35.000+02:00
Support responder checking for existence of requester via HTTP GET (over AMQP).
This is especially useful for when the responder attaches to the anonymous target.
This is in addition to attaching to the queues target with RabbitMQ
refusing the link if the queue doesn't exist.
diff --git a/deps/rabbit/src/rabbit_amqp_management.erl b/deps/rabbit/src/rabbit_amqp_management.erl
@@ -87,8 +87,14 @@ handle_http_req(<<"GET">>,
            QName,
            fun(Q) ->
                    {ok, NumMsgs, NumConsumers} = rabbit_amqqueue:stat(Q),
-                   RespPayload = encode_queue(Q, NumMsgs, NumConsumers),
-                   {ok, {<<"200">>, RespPayload, PermCaches}}
+                   case rabbit_volatile_queue:is(QNameBin) andalso
+                        not rabbit_volatile_queue:exists(QName) of
+                       true ->
+                           {error, not_found};
+                       false ->
+                           RespPayload = encode_queue(Q, NumMsgs, NumConsumers),
+                           {ok, {<<"200">>, RespPayload, PermCaches}}
+                   end
            end) of
         {ok, Result} ->
             Result;
diff --git a/deps/rabbit/src/rabbit_volatile_queue.erl b/deps/rabbit/src/rabbit_volatile_queue.erl
@@ -285,7 +285,7 @@ notify_decorators(_) ->
     ok.
 
 stat(_) ->
-    {ok, 0, 0}.
+    {ok, 0, 1}.
 
 format(_, _) ->
     [].
diff --git a/deps/rabbit/test/direct_reply_to_amqp_SUITE.erl b/deps/rabbit/test/direct_reply_to_amqp_SUITE.erl
@@ -96,8 +96,8 @@ responder_attaches_queue_target(Config) ->
     RequestQueue = atom_to_binary(?FUNCTION_NAME),
     AddrRequestQueue = rabbitmq_amqp_address:queue(RequestQueue),
 
-    {ConnResponder, SessionResponder, LinkPair} = init(Config),
-    {ok, _} = rabbitmq_amqp_client:declare_queue(LinkPair, RequestQueue, #{}),
+    {ConnResponder, SessionResponder, LinkPairResponder} = init(Config),
+    {ok, _} = rabbitmq_amqp_client:declare_queue(LinkPairResponder, RequestQueue, #{}),
 
     OpnConfRequester0 = connection_config(Config),
     OpnConfRequester = OpnConfRequester0#{container_id := <<"requester">>,
@@ -143,6 +143,14 @@ responder_attaches_queue_target(Config) ->
     %% The metadata store should store only the request queue.
     ?assertEqual(1, rabbit_ct_broker_helpers:rpc(Config, rabbit_db_queue, count, [])),
 
+    {ok, #{queue := ReplyQ}} = rabbitmq_amqp_address:to_map(ReplyToAddr),
+    ?assertMatch({ok, #{vhost := <<"/">>,
+                        durable := false,
+                        type := <<"rabbit_volatile_queue">>,
+                        message_count := 0,
+                        consumer_count := 1}},
+                 rabbitmq_amqp_client:get_queue(LinkPairResponder, ReplyQ)),
+
     {ok, SenderResponder1} = amqp10_client:attach_sender_link_sync(
                                SessionResponder, <<"sender responder unsettled">>,
                                ReplyToAddr, unsettled),
@@ -247,12 +255,15 @@ responder_attaches_queue_target(Config) ->
     end,
 
     %% When the requester detaches, the volatile queue is gone.
-    %% Therefore, RabbitMQ should refuse attaching to the volatile queue target.
     ok = detach_link_sync(ReceiverRequester),
     flush(detached),
     ?assertMatch(#{publishers := 3,
                    consumers := 1},
                  maps:get(#{protocol => amqp10}, get_global_counters(Config))),
+    %% Therefore, HTTP GET on that queue should return 404.
+    {error, Resp} = rabbitmq_amqp_client:get_queue(LinkPairResponder, ReplyQ),
+    ?assertMatch(#{subject := <<"404">>}, amqp10_msg:properties(Resp)),
+    %% Also, RabbitMQ should refuse attaching to the volatile queue target.
     {ok, SenderResponder3} = amqp10_client:attach_sender_link_sync(
                                SessionResponder, <<"sender responder 3">>,
                                ReplyToAddr),
@@ -278,8 +289,8 @@ responder_attaches_queue_target(Config) ->
     after 9000 -> ct:fail({missing_event, ?LINE})
     end,
 
-    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, RequestQueue),
-    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
+    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPairResponder, RequestQueue),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPairResponder),
 
     ok = close_connection_sync(ConnResponder),
     ok = close_connection_sync(ConnRequester),
diff --git a/deps/rabbit/test/direct_reply_to_amqpl_SUITE.erl b/deps/rabbit/test/direct_reply_to_amqpl_SUITE.erl
@@ -18,7 +18,9 @@
 -import(rabbit_ct_helpers,
         [eventually/1]).
 -import(amqp_utils,
-        [connection_config/1,
+        [init/1,
+         close/1,
+         connection_config/1,
          wait_for_credit/1,
          end_session_sync/1,
          close_connection_sync/1]).
@@ -300,16 +302,23 @@ amqpl_amqp_amqpl(Config) ->
                                 payload = RequestPayload}),
 
     %% Receive the request via AMQP 1.0.
-    OpnConf = connection_config(Config),
-    {ok, Conn} = amqp10_client:open_connection(OpnConf),
-    {ok, Session} = amqp10_client:begin_session_sync(Conn),
+    {_, Session, LinkPair} = Init = init(Config),
     {ok, Receiver} = amqp10_client:attach_receiver_link(
                        Session, <<"receiver">>, AddrRequestQ),
     {ok, RequestMsg} = amqp10_client:get_msg(Receiver, ?TIMEOUT),
     ?assertEqual(RequestPayload, amqp10_msg:body_bin(RequestMsg)),
     #{message_id := Id,
       reply_to := ReplyToAddr} = amqp10_msg:properties(RequestMsg),
 
+    %% AMQP 1.0 responder checks whether the AMQP 0.9.1 requester is still there.
+    {ok, #{queue := ReplyQ}} = rabbitmq_amqp_address:to_map(ReplyToAddr),
+    ?assertMatch({ok, #{vhost := <<"/">>,
+                        durable := false,
+                        type := <<"rabbit_volatile_queue">>,
+                        message_count := 0,
+                        consumer_count := 1}},
+                 rabbitmq_amqp_client:get_queue(LinkPair, ReplyQ)),
+
     %% Send the reply via AMQP 1.0.
     {ok, Sender} = amqp10_client:attach_sender_link_sync(
                      Session, <<"sender">>, ReplyToAddr),
@@ -346,9 +355,19 @@ amqpl_amqp_amqpl(Config) ->
     after ?TIMEOUT -> ct:fail(missing_reply)
     end,
 
-    ok = end_session_sync(Session),
-    ok = close_connection_sync(Conn),
+    %% AMQP 0.9.1 requester cancels consumption.
+    ?assertMatch(#'basic.cancel_ok'{consumer_tag = CTag},
+                 amqp_channel:call(Chan, #'basic.cancel'{consumer_tag = CTag})),
+
+    %% This time, when the AMQP 1.0 responder checks whether the AMQP 0.9.1 requester
+    %% is still there, an error should be returned.
+    {error, Resp} = rabbitmq_amqp_client:get_queue(LinkPair, ReplyQ),
+    ?assertMatch(#{subject := <<"404">>}, amqp10_msg:properties(Resp)),
+    ?assertEqual(#'v1_0.amqp_value'{content = {utf8, <<"queue '", ReplyQ/binary, "' in vhost '/' not found">>}},
+                 amqp10_msg:body(Resp)),
+
     #'queue.delete_ok'{} = amqp_channel:call(Chan, #'queue.delete'{queue = RequestQ}),
+    ok = close(Init),
     ok = rabbit_ct_client_helpers:close_channel(Chan).
 
 %% Test that Direct Reply-To works when the requester is an AMQP 1.0 client
diff --git a/deps/rabbitmq_amqp_client/src/rabbitmq_amqp_address.erl b/deps/rabbitmq_amqp_client/src/rabbitmq_amqp_address.erl
@@ -8,7 +8,13 @@
 
 -export([exchange/1,
          exchange/2,
-         queue/1]).
+         queue/1,
+         from_map/1,
+         to_map/1]).
+
+-type address_map() :: #{queue := unicode:unicode_binary()} |
+                       #{exchange := unicode:unicode_binary(),
+                         routing_key => unicode:unicode_binary()}.
 
 -spec exchange(unicode:unicode_binary()) ->
     unicode:unicode_binary().
@@ -28,3 +34,57 @@ exchange(ExchangeName, RoutingKey) ->
 queue(QueueName) ->
     QueueNameQuoted = uri_string:quote(QueueName),
     <<"/queues/", QueueNameQuoted/binary>>.
+
+-spec from_map(address_map()) ->
+    unicode:unicode_binary().
+from_map(#{exchange := Exchange, routing_key := RoutingKey}) ->
+    exchange(Exchange, RoutingKey);
+from_map(#{exchange := Exchange}) ->
+    exchange(Exchange);
+from_map(#{queue := Queue}) ->
+    queue(Queue).
+
+-spec to_map(unicode:unicode_binary()) ->
+    {ok, address_map()} | error.
+to_map(<<"/exchanges/", Rest/binary>>) ->
+    case binary:split(Rest, <<"/">>, [global]) of
+        [ExchangeQuoted]
+          when ExchangeQuoted =/= <<>> ->
+            Exchange = uri_string:unquote(ExchangeQuoted),
+            {ok, #{exchange => Exchange}};
+        [ExchangeQuoted, RoutingKeyQuoted]
+          when ExchangeQuoted =/= <<>> ->
+            Exchange = uri_string:unquote(ExchangeQuoted),
+            RoutingKey = uri_string:unquote(RoutingKeyQuoted),
+            {ok, #{exchange => Exchange,
+                   routing_key => RoutingKey}};
+        _ ->
+            error
+    end;
+to_map(<<"/queues/">>) ->
+    error;
+to_map(<<"/queues/", QueueQuoted/binary>>) ->
+    Queue = uri_string:unquote(QueueQuoted),
+    {ok, #{queue => Queue}};
+to_map(_) ->
+    error.
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+address_test() ->
+    M1 = #{queue => <<"my queue">>},
+    M2 = #{queue => <<"🥕"/utf8>>},
+    M3 = #{exchange => <<"my exchange">>},
+    M4 = #{exchange => <<"🥕"/utf8>>},
+    M5 = #{exchange => <<"my exchange">>,
+           routing_key => <<"my routing key">>},
+    M6 = #{exchange => <<"🥕"/utf8>>,
+           routing_key => <<"🍰"/utf8>>},
+    lists:foreach(fun(Map) ->
+                          {ok, Map} = to_map(from_map(Map))
+                  end, [M1, M2, M3, M4, M5, M6]),
+
+    error = to_map(<<"/queues/">>),
+    error = to_map(<<"/exchanges/">>),
+    error = to_map(<<"/exchanges//key">>).
+-endif.
