Unbind queue

ansd · ansd · commit a51e8ebaa386 · 2024-02-26T18:57:15.000+01:00
diff --git a/deps/amqp10_client/src/amqp10_client_types.erl b/deps/amqp10_client/src/amqp10_client_types.erl
@@ -64,7 +64,7 @@
                                 link_event_detail()}.
 -type amqp10_event() :: {amqp10_event, amqp10_event_detail()}.
 
--type properties() :: #{binary() => tuple()}.
+-type properties() :: #{binary() => amqp10_binary_generator:amqp10_prim()}.
 
 -export_type([amqp10_performative/0, channel/0,
               source/0, target/0, amqp10_msg_record/0,
diff --git a/deps/amqp10_common/src/amqp10_binary_generator.erl b/deps/amqp10_common/src/amqp10_binary_generator.erl
@@ -52,6 +52,7 @@
 -export_type([
               amqp10_ctor/0,
               amqp10_type/0,
+              amqp10_prim/0,
               amqp10_described/0
              ]).
 
diff --git a/deps/rabbit/src/mc_amqpl.erl b/deps/rabbit/src/mc_amqpl.erl
@@ -26,7 +26,8 @@
          message/4,
          message/5,
          from_basic_message/1,
-         to_091/2
+         to_091/2,
+         from_091/2
         ]).
 
 -import(rabbit_misc,
diff --git a/deps/rabbit/src/rabbit_amqp_management.erl b/deps/rabbit/src/rabbit_amqp_management.erl
@@ -186,21 +186,42 @@ process_http_request(<<"POST">>,
     {Props, AppProps, RespPayload};
 
 process_http_request(<<"DELETE">>,
-                     <<"/$management/queues/", QNameBin/binary>>,
+                     <<"/$management/queues/", Path/binary>>,
                      undefined,
                      [],
                      Vhost,
                      #user{username = Username},
                      ConnectionPid) ->
-    QName = rabbit_misc:r(Vhost, queue, QNameBin),
-    {ok, NumMsgs} = rabbit_amqqueue:delete_with(QName, ConnectionPid, false, false, Username, true),
-    Props = #'v1_0.properties'{
-               subject = {utf8, <<"200">>},
-               content_type = {symbol, <<"application/amqp-management+amqp">>}
-              },
-    AppProps = #'v1_0.application_properties'{content = []},
-    RespPayload = {map, [{{utf8, <<"message_count">>}, {ulong, NumMsgs}}]},
-    {Props, AppProps, RespPayload};
+    case re:split(Path, <<"/">>, [{return, binary}]) of
+        [QNameBin] ->
+            QName = rabbit_misc:r(Vhost, queue, QNameBin),
+            {ok, NumMsgs} = rabbit_amqqueue:delete_with(QName, ConnectionPid, false, false, Username, true),
+            Props = #'v1_0.properties'{
+                       subject = {utf8, <<"200">>},
+                       content_type = {symbol, <<"application/amqp-management+amqp">>}
+                      },
+            AppProps = #'v1_0.application_properties'{content = []},
+            RespPayload = {map, [{{utf8, <<"message_count">>}, {ulong, NumMsgs}}]},
+            {Props, AppProps, RespPayload};
+        [QNameBin, <<"bindings">>, SrcXNameBin, BindingKey, ArgsHash] ->
+            SrcXName = rabbit_misc:r(Vhost, exchange, SrcXNameBin),
+            QName = rabbit_misc:r(Vhost, queue, QNameBin),
+            Bindings0 = rabbit_binding:list_for_source_and_destination(SrcXName, QName, true),
+            case lists:search(fun(#binding{key = Key,
+                                           args = Args}) ->
+                                      Key =:= BindingKey andalso
+                                      args_hash(Args) =:= ArgsHash
+                              end, Bindings0) of
+                {value, Binding} ->
+                    ok = rabbit_binding:remove(Binding, Username);
+                false ->
+                    ok
+            end,
+            Props = #'v1_0.properties'{subject = {utf8, <<"204">>}},
+            AppProps = #'v1_0.application_properties'{content = []},
+            RespPayload = {map, []},
+            {Props, AppProps, RespPayload}
+    end;
 
 process_http_request(<<"DELETE">>,
                      <<"/$management/exchanges/", XNameBin/binary>>,
@@ -221,6 +242,31 @@ process_http_request(<<"DELETE">>,
     Props = #'v1_0.properties'{subject = {utf8, <<"204">>}},
     AppProps = #'v1_0.application_properties'{content = []},
     RespPayload = {map, []},
+    {Props, AppProps, RespPayload};
+
+process_http_request(<<"GET">>,
+                     <<"/$management/", Path0/binary>>,
+                     undefined,
+                     [],
+                     Vhost,
+                     _User,
+                     _) ->
+    {DstKind, Path} = case Path0 of
+                          <<"queues/", Path1/binary>> ->
+                              {queue, Path1};
+                          <<"exchanges/", Path1/binary>> ->
+                              {exchange, Path1}
+                      end,
+    [DstNameBin, SrcXNameBin] = re:split(Path, <<"/\\$management/bindings\\?source=">>, [{return, binary}]),
+    SrcXName = rabbit_misc:r(Vhost, exchange, SrcXNameBin),
+    DstName = rabbit_misc:r(Vhost, DstKind, DstNameBin),
+    Bindings0 = rabbit_binding:list_for_source_and_destination(SrcXName, DstName, true),
+    RespPayload = encode_bindings(Bindings0),
+    Props = #'v1_0.properties'{
+               subject = {utf8, <<"200">>},
+               content_type = {symbol, <<"application/amqp-management+amqp">>}
+              },
+    AppProps = #'v1_0.application_properties'{content = []},
     {Props, AppProps, RespPayload}.
 
 decode_entity({map, KVList}) ->
@@ -267,6 +313,38 @@ decode_binding({map, KVList}) ->
               Acc#{arguments => Args}
       end, #{}, KVList).
 
+encode_bindings(Bindings) ->
+    Bs = lists:map(
+           fun(#binding{source = #resource{name = SrcName},
+                        key = BindingKey,
+                        destination = #resource{kind = DstKind,
+                                                name = DstName},
+                        args = Args091}) ->
+                   DstKindBin = case DstKind of
+                                    queue -> <<"queue">>;
+                                    exchange -> <<"exchange">>
+                                end,
+                   Args = [{{utf8, Key}, mc_amqpl:from_091(Type, Val)}
+                           || {Key, Type, Val} <- Args091],
+                   ArgsHash = args_hash(Args091),
+                   Self = <<"/$management/",
+                            DstKindBin/binary, "s/",
+                            DstName/binary, "/bindings/",
+                            SrcName/binary, "/",
+                            BindingKey/binary, "/",
+                            ArgsHash/binary>>,
+                   KVList = [
+                             {{utf8, <<"type">>}, {utf8, <<"binding">>}},
+                             {{utf8, <<"source">>}, {utf8, SrcName}},
+                             {{utf8, DstKindBin}, {utf8, DstName}},
+                             {{utf8, <<"binding_key">>}, {utf8, BindingKey}},
+                             {{utf8, <<"arguments">>}, {map, Args}},
+                             {{utf8, <<"self">>}, {utf8, Self}}
+                            ],
+                   {map, KVList}
+           end, Bindings),
+    {list, Bs}.
+
 decode_req([], {Props, AppProps, DataRev}) ->
     #msg{properties = Props,
          application_properties = AppProps,
diff --git a/deps/rabbitmq_amqp_client/src/rabbitmq_amqp_client.erl b/deps/rabbitmq_amqp_client/src/rabbitmq_amqp_client.erl
@@ -13,7 +13,7 @@
         declare_exchange/2,
         bind_queue/5,
         bind_exchange/5,
-        % unbind_queue/5,
+        unbind_queue/5,
         % unbind_exchange/5,
         purge_queue/2,
         delete_queue/2,
@@ -42,6 +42,8 @@
                                  internal => boolean(),
                                  arguments => x_args()}.
 
+-type amqp10_prim() :: amqp10_binary_generator:amqp10_prim().
+
 -export_type([link_pair/0]).
 
 -spec attach_management_link_pair_sync(pid(), binary()) ->
@@ -140,7 +142,7 @@ declare_queue(#link_pair{outgoing_link = OutgoingLink,
         {ok, proplists:to_map(KVList)}
     end.
 
--spec bind_queue(link_pair(), binary(), binary(), binary(), #{binary() => {atom(), term()}}) ->
+-spec bind_queue(link_pair(), binary(), binary(), binary(), #{binary() => amqp10_prim()}) ->
     ok | {error, term()}.
 bind_queue(#link_pair{outgoing_link = OutgoingLink,
                       incoming_link = IncomingLink},
@@ -184,7 +186,7 @@ bind_queue(#link_pair{outgoing_link = OutgoingLink,
         ok
     end.
 
--spec bind_exchange(link_pair(), binary(), binary(), binary(), #{binary() => {atom(), term()}}) ->
+-spec bind_exchange(link_pair(), binary(), binary(), binary(), #{binary() => amqp10_prim()}) ->
     ok | {error, term()}.
 bind_exchange(#link_pair{outgoing_link = OutgoingLink,
                          incoming_link = IncomingLink},
@@ -228,6 +230,88 @@ bind_exchange(#link_pair{outgoing_link = OutgoingLink,
         ok
     end.
 
+-spec unbind_queue(link_pair(), binary(), binary(), binary(), #{binary() => amqp10_prim()}) ->
+    ok | {error, term()}.
+unbind_queue(LinkPair = #link_pair{outgoing_link = OutgoingLink,
+                                   incoming_link = IncomingLink},
+             QueueName, ExchangeName, BindingKey, BindingArguments) ->
+    MessageId = message_id(),
+    HttpMethod = <<"GET">>,
+    HttpRequestTarget = <<"/$management/queues/", QueueName/binary,
+                          "/$management/bindings?source=", ExchangeName/binary>>,
+    Props = #{message_id => {binary, MessageId},
+              to => HttpRequestTarget,
+              subject => HttpMethod,
+              reply_to => <<"$me">>},
+    Req0 = amqp10_msg:new(<<>>, <<>>, true),
+    Req = amqp10_msg:set_properties(Props, Req0),
+
+    ok = amqp10_client:flow_link_credit(IncomingLink, 1, never),
+    maybe
+        ok ?= amqp10_client:send_msg(OutgoingLink, Req),
+        {ok, Resp} ?= receive {amqp10_msg, IncomingLink, Message} -> {ok, Message}
+                      after ?TIMEOUT -> {error, response_timeout}
+                      end,
+        #{correlation_id := MessageId,
+          subject := <<"200">>,
+          content_type := <<"application/amqp-management+amqp">>
+         } = amqp10_msg:properties(Resp),
+        #{<<"http:response">> := <<"1.1">> } = amqp10_msg:application_properties(Resp),
+        RespBody = amqp10_msg:body_bin(Resp),
+        [{list, Bindings}] = amqp10_framing:decode_bin(RespBody),
+        case binding_uri(BindingKey, BindingArguments, Bindings) of
+            {ok, Uri} ->
+                ok = delete_queue_binding(LinkPair, Uri);
+            not_found ->
+                ok
+        end
+    end.
+
+binding_uri(_, _, []) ->
+    not_found;
+binding_uri(BindingKey, BindingArguments, [{map, KVList} | Bindings]) ->
+    case maps:from_list(KVList) of
+        #{{utf8, <<"binding_key">>} := {utf8, BindingKey},
+          {utf8, <<"arguments">>} := {map, Args},
+          {utf8, <<"self">>} := {utf8, Uri}} ->
+            Args = lists:map(fun({{utf8, Key}, TypeVal}) ->
+                                     {Key, TypeVal}
+                             end, Args),
+            case maps:from_list(Args) =:= BindingArguments of
+                true ->
+                    {ok, Uri};
+                false ->
+                    binding_uri(BindingKey, BindingArguments, Bindings)
+            end;
+        _ ->
+            binding_uri(BindingKey, BindingArguments, Bindings)
+    end.
+
+-spec delete_queue_binding(link_pair(), binary()) ->
+    ok | {error, term()}.
+delete_queue_binding(#link_pair{outgoing_link = OutgoingLink,
+                                incoming_link = IncomingLink}, BindingUri) ->
+    MessageId = message_id(),
+    HttpMethod = <<"DELETE">>,
+    Props = #{message_id => {binary, MessageId},
+              to => BindingUri,
+              subject => HttpMethod,
+              reply_to => <<"$me">>},
+    Req0 = amqp10_msg:new(<<>>, <<>>, true),
+    Req = amqp10_msg:set_properties(Props, Req0),
+
+    ok = amqp10_client:flow_link_credit(IncomingLink, 1, never),
+    maybe
+        ok ?= amqp10_client:send_msg(OutgoingLink, Req),
+        {ok, Resp} ?= receive {amqp10_msg, IncomingLink, Message} -> {ok, Message}
+                      after ?TIMEOUT -> {error, response_timeout}
+                      end,
+        #{correlation_id := MessageId,
+          subject := <<"204">>} = amqp10_msg:properties(Resp),
+        #{<<"http:response">> := <<"1.1">> } = amqp10_msg:application_properties(Resp),
+        ok
+    end.
+
 -spec purge_queue(link_pair(), binary()) ->
     {ok, map()} | {error, term()}.
 purge_queue(#link_pair{outgoing_link = OutgoingLink,
diff --git a/deps/rabbitmq_amqp_client/test/management_SUITE.erl b/deps/rabbitmq_amqp_client/test/management_SUITE.erl
@@ -95,6 +95,11 @@ all_management_operations(Config) ->
     ok = amqp10_client:send_msg(Sender2, Msg2),
     ok = wait_for_accepted(DTag2),
 
+    ?assertEqual(ok, rabbitmq_amqp_client:unbind_queue(LinkPair, QName, SourceExchange, BindingKey1, #{})),
+    DTag3 = <<"tag 3">>,
+    ok = amqp10_client:send_msg(Sender2, amqp10_msg:new(DTag3, <<"not routed">>, false)),
+    ok = wait_for_settlement(DTag3, released),
+
     XName = <<"my-fanout-exchange">>,
     X = #{name => XName,
           type => <<"fanout">>,
@@ -110,10 +115,10 @@ all_management_operations(Config) ->
     {ok, Sender3} = amqp10_client:attach_sender_link(Session, <<"sender 3">>, TargetAddr3),
     ok = wait_for_credit(Sender3),
     flush(credited),
-    DTag3 = <<"tag 3">>,
-    Msg3 = amqp10_msg:new(DTag3, <<"m3">>, false),
+    DTag4 = <<"tag 4">>,
+    Msg3 = amqp10_msg:new(DTag4, <<"m3">>, false),
     ok = amqp10_client:send_msg(Sender3, Msg3),
-    ok = wait_for_accepted(DTag3),
+    ok = wait_for_accepted(DTag4),
 
     RoutingKey2 = BindingKey2 = <<"key 2">>,
     ?assertEqual(ok, rabbitmq_amqp_client:bind_exchange(LinkPair, XName, SourceExchange, BindingKey2, #{})),
@@ -122,17 +127,17 @@ all_management_operations(Config) ->
     {ok, Sender4} = amqp10_client:attach_sender_link(Session, <<"sender 4">>, TargetAddr4),
     ok = wait_for_credit(Sender4),
     flush(credited),
-    DTag4 = <<"tag 4">>,
-    Msg4 = amqp10_msg:new(DTag4, <<"m4">>, false),
+    DTag5 = <<"tag 5">>,
+    Msg4 = amqp10_msg:new(DTag5, <<"m4">>, false),
     ok = amqp10_client:send_msg(Sender4, Msg4),
-    ok = wait_for_accepted(DTag4),
+    ok = wait_for_accepted(DTag5),
 
     ?assertEqual(ok, rabbitmq_amqp_client:delete_exchange(LinkPair, XName)),
     %% When we publish the next message, we expect:
     %% 1. that the message is released because the exchange doesn't exist anymore, and
-    DTag5 = <<"tag 5">>,
-    ok = amqp10_client:send_msg(Sender3, amqp10_msg:new(DTag5, <<"m5">>, false)),
-    ok = wait_for_settlement(DTag5, released),
+    DTag6 = <<"tag 6">>,
+    ok = amqp10_client:send_msg(Sender3, amqp10_msg:new(DTag6, <<"not routed">>, false)),
+    ok = wait_for_settlement(DTag6, released),
     %% 2. that the server closes the link, i.e. sends us a DETACH frame.
     ExpectedError = #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_RESOURCE_DELETED},
     receive {amqp10_event, {link, Sender3, {detached, ExpectedError}}} -> ok
