Bind queue

Author: ansd

deps/rabbit/src/rabbit_amqp_management.erl

@@ -4,6 +4,7 @@
 -include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([process_request/4]).
+-export([args_hash/1]).
 
 %% An HTTP message mapped to AMQP using projected mode
 %% [HTTP over AMQP Working Draft 06 §4.1]
@@ -84,6 +85,47 @@ process_http_request(<<"POST">>,
                         ]},
     {Props, AppProps, RespPayload};
 
+process_http_request(<<"POST">>,
+                     <<"/$management/queues/", QNamePath/binary>>,
+                     {symbol, <<"application/amqp-management+amqp", _OptionalType/binary>>},
+                     [ReqPayload],
+                     Vhost,
+                     #user{username = Username},
+                     _) ->
+    [QNameBin, <<>>] = re:split(QNamePath, <<"/\\$management/entities$">>, [{return, binary}]),
+    %% TODO add a top level binding key so that other entity types can be created in future by posting
+    %% to the queue's $management/entities URI
+    #{source_exchange := XNameBin,
+      binding_key := BindingKey,
+      arguments := Args} = decode_binding(ReqPayload),
+    XName = rabbit_misc:r(Vhost, exchange, XNameBin),
+    QName = rabbit_misc:r(Vhost, queue, QNameBin),
+    Binding = #binding{source      = XName,
+                       destination = QName,
+                       key         = BindingKey,
+                       args        = Args},
+    %%TODO If the binding already exists, return 303 with location.
+    ok = rabbit_binding:add(Binding, Username),
+    Props = #'v1_0.properties'{
+               subject = {utf8, <<"201">>},
+               content_type = {symbol, <<"application/amqp-management+amqp;type=entity-collection">>}
+              },
+    %% TODO URI encode, which is against the HTTP over AMQP spec [3.1].
+    %% How else to escape "/" in exchange names and binding keys?
+    ArgsHash = args_hash(Args),
+    Self = {utf8, <<"/$management/queues/",
+                    QNameBin/binary, "/bindings/",
+                    XNameBin/binary, "/",
+                    BindingKey/binary, "/",
+                    ArgsHash/binary>>},
+    AppProps = #'v1_0.application_properties'{
+                  content = [{{utf8, <<"location">>}, Self}]},
+    %% TODO Include source_exchange, binding key, and binding arguments in the response?
+    RespPayload = {map, [{{utf8, <<"type">>}, {utf8, <<"binding">>}},
+                         {{utf8, <<"self">>}, Self}
+                        ]},
+    {Props, AppProps, RespPayload};
+
 process_http_request(<<"POST">>,
                      <<"/$management/queues/", QNamePath/binary>>,
                      undefined,
@@ -149,6 +191,18 @@ decode_queue({map, KVList}) ->
               Acc#{arguments => Args}
       end, #{}, KVList).
 
+decode_binding({map, KVList}) ->
+    lists:foldl(
+      fun({{utf8, <<"source_exchange">>}, {utf8, V}}, Acc) ->
+              Acc#{source_exchange => V};
+         ({{utf8, <<"binding_key">>}, {utf8, V}}, Acc) ->
+              Acc#{binding_key => V};
+         ({{utf8, <<"arguments">>}, {map, List}}, Acc) ->
+              Args = [mc_amqpl:to_091(Key, TypeVal)
+                      || {{utf8, Key}, TypeVal} <- List],
+              Acc#{arguments => Args}
+      end, #{}, KVList).
+
 decode_req([], {Props, AppProps, DataRev}) ->
     #msg{properties = Props,
          application_properties = AppProps,
@@ -161,3 +215,7 @@ decode_req([#'v1_0.data'{content = C} | Rem], {Props, AppProps, DataRev}) ->
     decode_req(Rem, {Props, AppProps, [C | DataRev]});
 decode_req([_IgnoreOtherSection | Rem], Acc) ->
     decode_req(Rem, Acc).
+
+-spec args_hash(rabbit_framing:amqp_table()) -> binary().
+args_hash(Args) ->
+    list_to_binary(rabbit_misc:base64url(<<(erlang:phash2(Args, 1 bsl 32)):32>>)).

deps/rabbitmq_amqp_client/src/rabbitmq_amqp_client.erl

@@ -10,6 +10,8 @@
 
 -export[attach_management_link_pair_sync/2,
         declare_queue/2,
+        bind_queue/5,
+        % unbind_queue/5,
         purge_queue/2,
         delete_queue/2].
 
@@ -126,6 +128,50 @@ declare_queue(#link_pair{outgoing_link = OutgoingLink,
         {ok, proplists:to_map(KVList)}
     end.
 
+-spec bind_queue(link_pair(), binary(), binary(), binary(), #{binary() => {atom(), term()}}) ->
+    ok | {error, term()}.
+bind_queue(#link_pair{outgoing_link = OutgoingLink,
+                      incoming_link = IncomingLink},
+           QueueName, ExchangeName, BindingKey, BindingArguments) ->
+    KVList = maps:fold(
+               fun(Key, TaggedVal = {T, _}, L)
+                     when is_binary(Key) andalso is_atom(T) ->
+                       [{{utf8, Key}, TaggedVal} | L]
+               end, [], BindingArguments),
+    Body0 = {map, [
+                   {{utf8, <<"source_exchange">>}, {utf8, ExchangeName}},
+                   {{utf8, <<"binding_key">>}, {utf8, BindingKey}},
+                   {{utf8, <<"arguments">>}, {map, KVList}}
+                  ]},
+    Body = iolist_to_binary(amqp10_framing:encode_bin(Body0)),
+
+    MessageId = message_id(),
+    HttpMethod = <<"POST">>,
+    HttpRequestTarget = <<"/$management/queues/", QueueName/binary, "/$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">>
+         } = amqp10_msg:properties(Resp),
+        #{<<"http:response">> := <<"1.1">>,
+          <<"location">> := _BindingRUI
+         } = amqp10_msg:application_properties(Resp),
+        ok
+    end.
+
 -spec purge_queue(link_pair(), binary()) ->
     {ok, map()} | {error, term()}.
 purge_queue(#link_pair{outgoing_link = OutgoingLink,

deps/rabbitmq_amqp_client/test/management_SUITE.erl

@@ -72,17 +72,30 @@ queue(Config) ->
           exclusive => false,
           auto_delete => false,
           arguments => #{<<"x-queue-type">> => {symbol, <<"quorum">>}}},
-    {ok, #{{utf8, <<"target">>} := {utf8, Address}} } = rabbitmq_amqp_client:declare_queue(LinkPair, Q),
+    {ok, #{{utf8, <<"target">>} := {utf8, TargetAddr1}} } = rabbitmq_amqp_client:declare_queue(LinkPair, Q),
 
-    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"test-sender">>, Address),
-    ok = wait_for_credit(Sender),
+    {ok, Sender1} = amqp10_client:attach_sender_link(Session, <<"sender 1">>, TargetAddr1),
+    ok = wait_for_credit(Sender1),
     flush(credited),
-    DTag = <<"tag 1">>,
-    Msg1 = amqp10_msg:new(DTag, <<"m1">>, false),
-    ok = amqp10_client:send_msg(Sender, Msg1),
-    ok = wait_for_accepted(DTag),
+    DTag1 = <<"tag 1">>,
+    Msg1 = amqp10_msg:new(DTag1, <<"m1">>, false),
+    ok = amqp10_client:send_msg(Sender1, Msg1),
+    ok = wait_for_accepted(DTag1),
+
+    RoutingKey = BindingKey = <<"key 1">>,
+    SourceExchange = <<"amq.direct">>,
+    ?assertEqual(ok, rabbitmq_amqp_client:bind_queue(LinkPair, QName, SourceExchange, BindingKey, #{})),
+    TargetAddr2 = <<"/exchange/", SourceExchange/binary, "/", RoutingKey/binary>>,
+
+    {ok, Sender2} = amqp10_client:attach_sender_link(Session, <<"sender 2">>, TargetAddr2),
+    ok = wait_for_credit(Sender2),
+    flush(credited),
+    DTag2 = <<"tag 2">>,
+    Msg2 = amqp10_msg:new(DTag2, <<"m2">>, false),
+    ok = amqp10_client:send_msg(Sender2, Msg2),
+    ok = wait_for_accepted(DTag2),
 
-    ?assertEqual({ok, #{message_count => 1}},
+    ?assertEqual({ok, #{message_count => 2}},
                  rabbitmq_amqp_client:purge_queue(LinkPair, QName)),
 
     ?assertEqual({ok, #{message_count => 0}},

deps/rabbitmq_management/src/rabbit_mgmt_wm_binding.erl

@@ -127,7 +127,7 @@ lookup(RoutingKey, Hash, [#binding{args = Args} | Rest]) ->
     end.
 
 args_hash(Args) ->
-    rabbit_mgmt_format:args_hash(Args).
+    rabbit_amqp_management:args_hash(Args).
 
 unquote(Name) ->
     list_to_binary(rabbit_http_util:unquote(Name)).

deps/rabbitmq_management/test/rabbit_mgmt_http_SUITE.erl

@@ -2094,7 +2094,7 @@ arguments_test(Config) ->
     passed.
 
 table_hash(Table) ->
-    binary_to_list(rabbit_mgmt_format:args_hash(Table)).
+    binary_to_list(rabbit_amqp_management:args_hash(Table)).
 
 arguments_table_test(Config) ->
     Args = #{'upstreams' => [<<"amqp://localhost/%2F/upstream1">>,

deps/rabbitmq_management/test/rabbit_mgmt_only_http_SUITE.erl

@@ -892,7 +892,7 @@ arguments_test(Config) ->
     passed.
 
 table_hash(Table) ->
-    binary_to_list(rabbit_mgmt_format:args_hash(Table)).
+    binary_to_list(rabbit_amqp_management:args_hash(Table)).
 
 queue_actions_test(Config) ->
     http_put(Config, "/queues/%2F/q", #{}, {group, '2xx'}),

deps/rabbitmq_management_agent/src/rabbit_mgmt_format.erl

@@ -28,8 +28,6 @@
 
 -export([clean_consumer_details/1, clean_channel_details/1]).
 
--export([args_hash/1]).
-
 -import(rabbit_misc, [pget/2, pget/3, pset/3]).
 -import(rabbit_data_coercion, [to_binary/1]).
 
@@ -340,7 +338,7 @@ pack_binding_props(<<"">>, []) ->
 pack_binding_props(Key, []) ->
     list_to_binary(quote_binding(Key));
 pack_binding_props(Key, Args) ->
-    ArgsEnc = args_hash(Args),
+    ArgsEnc = rabbit_amqp_management:args_hash(Args),
     list_to_binary(quote_binding(Key) ++ "~" ++ quote_binding(ArgsEnc)).
 
 quote_binding(Name) ->
@@ -595,6 +593,3 @@ parse_bool(true)        -> true;
 parse_bool(false)       -> false;
 parse_bool(undefined)   -> undefined;
 parse_bool(V)           -> throw({error, {not_boolean, V}}).
-
-args_hash(Args) ->
-    list_to_binary(rabbit_misc:base64url(<<(erlang:phash2(Args, 1 bsl 32)):32>>)).