unbind exchange in AMQP Erlang client

Author: ansd

deps/rabbit/src/rabbit_amqp_management.erl

@@ -245,10 +245,10 @@ process_http_request(<<"DELETE">>,
             AppProps = #'v1_0.application_properties'{content = []},
             RespPayload = {map, []},
             {Props, AppProps, RespPayload};
-        [XNameBin, <<"bindings">>, SrcXNameBin, BindingKey, ArgsHash] ->
+        [DstXNameBin, <<"bindings">>, SrcXNameBin, BindingKey, ArgsHash] ->
             SrcXName = rabbit_misc:r(Vhost, exchange, SrcXNameBin),
-            XName = rabbit_misc:r(Vhost, exchange, XNameBin),
-            Bindings0 = rabbit_binding:list_for_source_and_destination(SrcXName, XName, true),
+            DstXName = rabbit_misc:r(Vhost, exchange, DstXNameBin),
+            Bindings0 = rabbit_binding:list_for_source_and_destination(SrcXName, DstXName, true),
             case lists:search(fun(#binding{key = Key,
                                            args = Args}) ->
                                       Key =:= BindingKey andalso

deps/rabbitmq_amqp_client/src/rabbitmq_amqp_client.erl

@@ -14,7 +14,7 @@
         bind_queue/5,
         bind_exchange/5,
         unbind_queue/5,
-        % unbind_exchange/5,
+        unbind_exchange/5,
         purge_queue/2,
         delete_queue/2,
         delete_exchange/2
@@ -232,13 +232,26 @@ bind_exchange(#link_pair{outgoing_link = OutgoingLink,
 
 -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) ->
+unbind_queue(LinkPair, QueueName, ExchangeName, BindingKey, BindingArguments) ->
+    unbind(<<"queues">>, LinkPair, QueueName, ExchangeName, BindingKey, BindingArguments).
+
+-spec unbind_exchange(link_pair(), binary(), binary(), binary(), #{binary() => amqp10_prim()}) ->
+    ok | {error, term()}.
+unbind_exchange(LinkPair, DestinationExchange, SourceExchange, BindingKey, BindingArguments) ->
+    unbind(<<"exchanges">>, LinkPair, DestinationExchange, SourceExchange, BindingKey, BindingArguments).
+
+-spec unbind(binary(), link_pair(), binary(), binary(), binary(), #{binary() => amqp10_prim()}) ->
+    ok | {error, term()}.
+unbind(Type,
+       #link_pair{outgoing_link = OutgoingLink,
+                  incoming_link = IncomingLink} = LinkPair,
+       Destination, Source, BindingKey, BindingArguments) ->
     MessageId = message_id(),
     HttpMethod = <<"GET">>,
-    HttpRequestTarget = <<"/$management/queues/", QueueName/binary,
-                          "/$management/bindings?source=", ExchangeName/binary>>,
+    HttpRequestTarget = <<"/$management/",
+                          Type/binary, "/",
+                          Destination/binary,
+                          "/$management/bindings?source=", Source/binary>>,
     Props = #{message_id => {binary, MessageId},
               to => HttpRequestTarget,
               subject => HttpMethod,
@@ -261,7 +274,7 @@ unbind_queue(LinkPair = #link_pair{outgoing_link = OutgoingLink,
         [{list, Bindings}] = amqp10_framing:decode_bin(RespBody),
         case binding_uri(BindingKey, BindingArguments, Bindings) of
             {ok, Uri} ->
-                ok = delete_queue_binding(LinkPair, Uri);
+                ok = delete_binding(LinkPair, Uri);
             not_found ->
                 ok
         end
@@ -287,10 +300,10 @@ binding_uri(BindingKey, BindingArguments, [{map, KVList} | Bindings]) ->
             binding_uri(BindingKey, BindingArguments, Bindings)
     end.
 
--spec delete_queue_binding(link_pair(), binary()) ->
+-spec delete_binding(link_pair(), binary()) ->
     ok | {error, term()}.
-delete_queue_binding(#link_pair{outgoing_link = OutgoingLink,
-                                incoming_link = IncomingLink}, BindingUri) ->
+delete_binding(#link_pair{outgoing_link = OutgoingLink,
+                          incoming_link = IncomingLink}, BindingUri) ->
     MessageId = message_id(),
     HttpMethod = <<"DELETE">>,
     Props = #{message_id => {binary, MessageId},

deps/rabbitmq_amqp_client/test/management_SUITE.erl

@@ -132,12 +132,17 @@ all_management_operations(Config) ->
     ok = amqp10_client:send_msg(Sender4, Msg4),
     ok = wait_for_accepted(DTag5),
 
+    ?assertEqual(ok, rabbitmq_amqp_client:unbind_exchange(LinkPair, XName, SourceExchange, BindingKey2, #{})),
+    DTag6 = <<"tag 6">>,
+    ok = amqp10_client:send_msg(Sender4, amqp10_msg:new(DTag6, <<"not routed">>, false)),
+    ok = wait_for_settlement(DTag6, released),
+
     ?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
-    DTag6 = <<"tag 6">>,
-    ok = amqp10_client:send_msg(Sender3, amqp10_msg:new(DTag6, <<"not routed">>, false)),
-    ok = wait_for_settlement(DTag6, released),
+    DTag7 = <<"tag 7">>,
+    ok = amqp10_client:send_msg(Sender3, amqp10_msg:new(DTag7, <<"not routed">>, false)),
+    ok = wait_for_settlement(DTag7, 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