Declare + delete exchange

ansd · ansd · commit 02e568c4db8c · 2024-02-25T17:08:36.000+01:00
diff --git a/deps/rabbit/src/rabbit_amqp_management.erl b/deps/rabbit/src/rabbit_amqp_management.erl
@@ -59,28 +59,55 @@ process_http_request(<<"POST">>,
                      Vhost,
                      #user{username = Username},
                      _) ->
-    #{name := QNameBin,
-      durable := Durable,
-      auto_delete := AutoDelete,
-      owner := Owner,
-      arguments := QArgs} = decode_queue(ReqPayload),
-    QType = rabbit_amqqueue:get_queue_type(QArgs),
-    QName = rabbit_misc:r(Vhost, queue, QNameBin),
-    Q0 = amqqueue:new(QName, none, Durable, AutoDelete, Owner,
-                      QArgs, Vhost, #{user => Username}, QType),
-    {new, _Q} = rabbit_queue_type:declare(Q0, node()),
+    {Type, Id, Self1, Target} =
+    case decode_entity(ReqPayload) of
+        #{type := <<"queue">> = Type0,
+          name := QNameBin,
+          durable := Durable,
+          auto_delete := AutoDelete,
+          owner := Owner,
+          arguments := QArgs} ->
+            QType = rabbit_amqqueue:get_queue_type(QArgs),
+            QName = rabbit_misc:r(Vhost, queue, QNameBin),
+            Q0 = amqqueue:new(QName, none, Durable, AutoDelete, Owner,
+                              QArgs, Vhost, #{user => Username}, QType),
+            {new, _Q} = rabbit_queue_type:declare(Q0, node()),
+            Self0 = <<"/$management/queues/", QNameBin/binary>>,
+            Target0 = <<"/queue/", QNameBin/binary>>,
+            {Type0, QNameBin, Self0, Target0};
+        #{type := <<"exchange">> = Type0,
+          name := XNameBin,
+          exchange_type := XType,
+          durable := Durable,
+          auto_delete := AutoDelete,
+          internal := Internal,
+          arguments := XArgs} ->
+            XName = rabbit_misc:r(Vhost, exchange, XNameBin),
+            CheckedXType = rabbit_exchange:check_type(XType),
+            _X = rabbit_exchange:declare(XName,
+                                         CheckedXType,
+                                         Durable,
+                                         AutoDelete,
+                                         Internal,
+                                         XArgs,
+                                         Username),
+            Self0 = <<"/$management/exchanges/", XNameBin/binary>>,
+            Target0 = <<"/exchange/", XNameBin/binary>>,
+            {Type0, XNameBin, Self0, Target0}
+
+    end,
+    Self = {utf8, Self1},
     Props = #'v1_0.properties'{
                subject = {utf8, <<"201">>},
                content_type = {symbol, <<"application/amqp-management+amqp;type=entity-collection">>}
               },
-    Self = {utf8, <<"/$management/queues/", QNameBin/binary>>},
     AppProps = #'v1_0.application_properties'{
                   %% TODO include vhost in URI?
                   content = [{{utf8, <<"location">>}, Self}]},
-    RespPayload = {map, [{{utf8, <<"type">>}, {utf8, <<"queue">>}},
-                         {{utf8, <<"id">>}, {utf8, QNameBin}},
+    RespPayload = {map, [{{utf8, <<"type">>}, {utf8, Type}},
+                         {{utf8, <<"id">>}, {utf8, Id}},
                          {{utf8, <<"self">>}, Self},
-                         {{utf8, <<"target">>}, {utf8, <<"/queue/", QNameBin/binary>>}},
+                         {{utf8, <<"target">>}, {utf8, Target}},
                          {{utf8, <<"management">>}, {utf8, <<"$management">>}}
                         ]},
     {Props, AppProps, RespPayload};
@@ -164,12 +191,33 @@ process_http_request(<<"DELETE">>,
               },
     AppProps = #'v1_0.application_properties'{content = []},
     RespPayload = {map, [{{utf8, <<"message_count">>}, {ulong, NumMsgs}}]},
+    {Props, AppProps, RespPayload};
+
+process_http_request(<<"DELETE">>,
+                     <<"/$management/exchanges/", XNameBin/binary>>,
+                     undefined,
+                     [],
+                     Vhost,
+                     #user{username = Username},
+                     _ConnectionPid) ->
+    XName = rabbit_misc:r(Vhost, exchange, XNameBin),
+    ok = case rabbit_exchange:delete(XName, false, Username) of
+             ok ->
+                 ok;
+             {error, not_found} ->
+                 ok
+                 %% %% TODO return deletion failure
+                 %% {error, in_use} ->
+         end,
+    Props = #'v1_0.properties'{subject = {utf8, <<"204">>}},
+    AppProps = #'v1_0.application_properties'{content = []},
+    RespPayload = {map, []},
     {Props, AppProps, RespPayload}.
 
-decode_queue({map, KVList}) ->
+decode_entity({map, KVList}) ->
     lists:foldl(
-      fun({{utf8, <<"type">>}, {utf8, <<"queue">>}}, Acc) ->
-              Acc;
+      fun({{utf8, <<"type">>}, {utf8, V}}, Acc) ->
+              Acc#{type => V};
          ({{utf8, <<"name">>}, {utf8, V}}, Acc) ->
               Acc#{name => V};
          ({{utf8, <<"durable">>}, V}, Acc)
@@ -184,6 +232,11 @@ decode_queue({map, KVList}) ->
          ({{utf8, <<"auto_delete">>}, V}, Acc)
            when is_boolean(V) ->
               Acc#{auto_delete => V};
+         ({{utf8, <<"exchange_type">>}, {utf8, V}}, Acc) ->
+              Acc#{exchange_type => V};
+         ({{utf8, <<"internal">>}, V}, Acc)
+           when is_boolean(V) ->
+              Acc#{internal => V};
          ({{utf8, <<"arguments">>}, {map, List}}, Acc) ->
               Args = [{Key, longstr, V}
                       || {{utf8, Key = <<"x-", _/binary>>},
diff --git a/deps/rabbit/src/rabbit_exchange.erl b/deps/rabbit/src/rabbit_exchange.erl
@@ -94,7 +94,7 @@ serial(X) ->
 -spec declare
         (name(), type(), boolean(), boolean(), boolean(),
          rabbit_framing:amqp_table(), rabbit_types:username())
-        -> rabbit_types:exchange().
+        -> rabbit_types:exchange() | {error, term()}.
 
 declare(XName, Type, Durable, AutoDelete, Internal, Args, Username) ->
     X = rabbit_exchange_decorator:set(
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,10 @@
         bind_queue/5,
         % unbind_queue/5,
         purge_queue/2,
-        delete_queue/2].
+        delete_queue/2,
+        declare_exchange/2,
+        delete_exchange/2
+       ].
 
 -define(TIMEOUT, 10_000).
 -define(MANAGEMENT_NODE_ADDRESS, <<"$management">>).
@@ -30,6 +33,13 @@
                               auto_delete => boolean(),
                               arguments => x_args()}.
 
+-type exchange_properties() :: #{name := binary(),
+                                 type => binary(),
+                                 durable => boolean(),
+                                 auto_delete => boolean(),
+                                 internal => boolean(),
+                                 arguments => x_args()}.
+
 -export_type([link_pair/0]).
 
 -spec attach_management_link_pair_sync(pid(), binary()) ->
@@ -240,6 +250,91 @@ delete_queue(#link_pair{outgoing_link = OutgoingLink,
         {ok, #{message_count => Count}}
     end.
 
+-spec declare_exchange(link_pair(), exchange_properties()) ->
+    {ok, map()} | {error, term()}.
+declare_exchange(#link_pair{outgoing_link = OutgoingLink,
+                            incoming_link = IncomingLink},
+                 ExchangeProperties) ->
+    Body0 = maps:fold(
+              fun(name, V, Acc) when is_binary(V) ->
+                      [{{utf8, <<"name">>}, {utf8, V}} | Acc];
+                 (type, V, Acc) when is_binary(V) ->
+                      [{{utf8, <<"exchange_type">>}, {utf8, V}} | Acc];
+                 (durable, V, Acc) when is_boolean(V) ->
+                      [{{utf8, <<"durable">>}, {boolean, V}} | Acc];
+                 (auto_delete, V, Acc) when is_boolean(V) ->
+                      [{{utf8, <<"auto_delete">>}, {boolean, V}} | Acc];
+                 (internal, V, Acc) when is_boolean(V) ->
+                      [{{utf8, <<"internal">>}, {boolean, V}} | Acc];
+                 (arguments, V, Acc) ->
+                      KVList = maps:fold(
+                                 fun(K = <<"x-", _/binary>>, TaggedVal = {T, _}, L)
+                                       when is_atom(T) ->
+                                         [{{utf8, K}, TaggedVal} | L]
+                                 end, [], V),
+                      [{{utf8, <<"arguments">>}, {map, KVList}} | Acc]
+              end, [{{utf8, <<"type">>}, {utf8, <<"exchange">>}}], ExchangeProperties),
+    Body1 = {map, Body0},
+    Body = iolist_to_binary(amqp10_framing:encode_bin(Body1)),
+
+    MessageId = message_id(),
+    HttpMethod = <<"POST">>,
+    HttpRequestTarget = <<"/$management/entities">>,
+    ContentType = <<"application/amqp-management+amqp;type=entity">>,
+    Props = #{message_id => {binary, MessageId},
+              to => HttpRequestTarget,
+              subject => HttpMethod,
+              reply_to => <<"$me">>,
+              content_type => ContentType},
+    Req0 = amqp10_msg:new(<<>>, Body, 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 := <<"201">>,
+          content_type := <<"application/amqp-management+amqp;type=entity-collection">>
+         } = amqp10_msg:properties(Resp),
+        #{<<"http:response">> := <<"1.1">>,
+          <<"location">> := _ExchangeURI
+         } = amqp10_msg:application_properties(Resp),
+        RespBody = amqp10_msg:body_bin(Resp),
+        [{map, KVList}] = amqp10_framing:decode_bin(RespBody),
+        {ok, proplists:to_map(KVList)}
+    end.
+
+-spec delete_exchange(link_pair(), binary()) ->
+    ok | {error, term()}.
+delete_exchange(#link_pair{outgoing_link = OutgoingLink,
+                           incoming_link = IncomingLink},
+                ExchangeName) ->
+    MessageId = message_id(),
+    HttpMethod = <<"DELETE">>,
+    HttpRequestTarget = <<"/$management/exchanges/", 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 := <<"204">>
+         } = amqp10_msg:properties(Resp),
+        #{<<"http:response">> := <<"1.1">> } = amqp10_msg:application_properties(Resp),
+        ok
+    end.
+
 %% "The message producer is usually responsible for setting the message-id in
 %% such a way that it is assured to be globally unique." [3.2.4]
 -spec message_id() -> binary().
diff --git a/deps/rabbitmq_amqp_client/test/management_SUITE.erl b/deps/rabbitmq_amqp_client/test/management_SUITE.erl
@@ -24,7 +24,7 @@ all() ->
 
 groups() ->
     [
-     {tests, [], [queue]}
+     {tests, [shuffle], [all_operations]}
     ].
 
 init_per_suite(Config) ->
@@ -55,7 +55,7 @@ init_per_testcase(Testcase, Config) ->
 end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
-queue(Config) ->
+all_operations(Config) ->
     OpnConf = connection_config(Config),
     {ok, Connection} = amqp10_client:open_connection(OpnConf),
     receive {amqp10_event, {connection, C, opened}}
@@ -95,7 +95,39 @@ queue(Config) ->
     ok = amqp10_client:send_msg(Sender2, Msg2),
     ok = wait_for_accepted(DTag2),
 
-    ?assertEqual({ok, #{message_count => 2}},
+    XName = <<"my-fanout-exchange">>,
+    X = #{name => XName,
+          type => <<"fanout">>,
+          durable => true,
+          auto_delete => false,
+          internal => false,
+          arguments => #{}},
+    {ok, #{{utf8, <<"target">>} := {utf8, TargetAddr3}} } = rabbitmq_amqp_client:declare_exchange(LinkPair, X),
+
+    SourceExchange = <<"amq.direct">>,
+    ?assertEqual(ok, rabbitmq_amqp_client:bind_queue(LinkPair, QName, XName, <<"ignored">>, #{})),
+
+    {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),
+    ok = amqp10_client:send_msg(Sender3, Msg3),
+    ok = wait_for_accepted(DTag3),
+
+    ?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
+    DTag4 = <<"tag 4">>,
+    ok = amqp10_client:send_msg(Sender3, amqp10_msg:new(DTag4, <<"m4">>, false)),
+    ok = wait_for_settlement(DTag4, 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
+    after 5000 -> ct:fail({missing_event, ?LINE})
+    end,
+
+    ?assertEqual({ok, #{message_count => 3}},
                  rabbitmq_amqp_client:purge_queue(LinkPair, QName)),
 
     ?assertEqual({ok, #{message_count => 0}},
