Use HTTP API v2

Author: ansd

deps/rabbit/src/rabbit.erl

@@ -1709,7 +1709,15 @@ persist_static_configuration() ->
                      _ ->
                          ?MAX_MSG_SIZE
                  end,
-    ok = persistent_term:put(max_message_size, MaxMsgSize).
+    ok = persistent_term:put(max_message_size, MaxMsgSize),
+
+    %% This regex matches for example
+    %% src=e1;dstq=q2;key=my-key;args=
+    %% Source, destination and binding key values must be percent encoded.
+    %% Binding args use the URL safe Base 64 Alphabet: https://datatracker.ietf.org/doc/html/rfc4648#section-5
+    {ok, MP} = re:compile(
+                 <<"^src=([0-9A-Za-z\-.\_\~%]+);dst([eq])=([0-9A-Za-z\-.\_\~%]+);key=([0-9A-Za-z\-.\_\~%]*);args=([0-9A-Za-z\-\_]*)$">>),
+    ok = persistent_term:put(mp_binding_uri_path_segment, MP).
 
 persist_static_configuration(Params) ->
     App = ?MODULE,

deps/rabbit/src/rabbit_amqp_management.erl

@@ -491,8 +491,7 @@ handle_1_0_connection_frame(
                         %% "the value in idle-time-out SHOULD be half the peer's actual timeout threshold" [2.4.5]
                         idle_time_out  = {uint, ReceiveTimeoutMillis div 2},
                         container_id   = {utf8, rabbit_nodes:cluster_name()},
-                        offered_capabilities = {array, symbol, [{symbol, <<"AMQP_MANAGEMENT_V1_0">>},
-                                                                {symbol, <<"LINK_PAIR_V1_0">>}]},
+                        offered_capabilities = {array, symbol, [{symbol, <<"LINK_PAIR_V1_0">>}]},
                         properties     = server_properties()}),
     State;
 handle_1_0_connection_frame(#'v1_0.close'{}, State0) ->

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -49,6 +49,14 @@
 %% we grant depending on how fast target queue(s) actually confirm messages.
 -define(LINK_CREDIT_RCV, 128).
 
+%% Pure HTTP clients of a future HTTP API (v2) would call endpoints as follows:
+%% GET /v2/vhosts/:vhost/queues/:queue
+%%
+%% Here, we use the terminus address /management/v2 so that AMQP 1.0 clients declare the HTTP API version
+%% at link attachment time. The vhost is already determined at AMQP connection open time.
+%% Therefore, there is no need to send the HTTP API version and the vhost in every HTTP over AMQP request.
+-define(MANAGEMENT_NODE_ADDRESS, <<"/management/v2">>).
+
 -export([start_link/8,
          process_frame/2,
          list_local/0,
@@ -735,7 +743,7 @@ handle_control(#'v1_0.attach'{
                   name = Name = {utf8, LinkName},
                   handle = Handle = ?UINT(HandleInt),
                   source = Source = #'v1_0.source'{address = ClientTerminusAddress},
-                  target = Target = #'v1_0.target'{address = {utf8, <<"$management">>}},
+                  target = Target = #'v1_0.target'{address = {utf8, ?MANAGEMENT_NODE_ADDRESS}},
                   initial_delivery_count = DeliveryCount = ?UINT(DeliveryCountInt),
                   properties = Properties
                  } = Attach,
@@ -792,7 +800,7 @@ handle_control(#'v1_0.attach'{
                   role = ?AMQP_ROLE_RECEIVER,
                   name = Name = {utf8, LinkName},
                   handle = Handle = ?UINT(HandleInt),
-                  source = Source = #'v1_0.source'{address = {utf8, <<"$management">>}},
+                  source = Source = #'v1_0.source'{address = {utf8, ?MANAGEMENT_NODE_ADDRESS}},
                   target = Target = #'v1_0.target'{address = ClientTerminusAddress},
                   rcv_settle_mode = RcvSettleMode,
                   max_message_size = MaybeMaxMessageSize,
@@ -1680,7 +1688,7 @@ incoming_mgmt_link_transfer(
     Settled = default(MaybeSettled, false),
     validate_transfer_rcv_settle_mode(RcvSettleMode, Settled),
     validate_message_size(Request, IncomingMaxMessageSize),
-    Response = rabbit_amqp_management:process_request(Request, Vhost, User, ReaderPid),
+    Response = rabbit_amqp_management:handle_request(Request, Vhost, User, ReaderPid),
 
     Transfer = #'v1_0.transfer'{
                   handle = ?UINT(OutgoingHandleInt),
@@ -1690,7 +1698,7 @@ incoming_mgmt_link_transfer(
                   settled = true},
     ?DEBUG("~s Outbound content:~n  ~tp~n",
            [?MODULE, [amqp10_framing:pprint(Section) ||
-                      Section <- amqp10_framing:decode_bin(iolist_to_binary(Respon))]]),
+                      Section <- amqp10_framing:decode_bin(iolist_to_binary(Response))]]),
     validate_message_size(Response, OutgoingMaxMessageSize),
     Frames = transfer_frames(Transfer, Response, MaxFrameSize),
     PendingTransfer = #pending_management_transfer{frames = Frames},
@@ -1917,7 +1925,7 @@ ensure_target(#'v1_0.target'{address = Address,
                 {ok, Dest} ->
                     QNameBin = ensure_terminus(target, Dest, Vhost, User, Durable),
                     {XNameList1, RK} = rabbit_routing_parser:parse_routing(Dest),
-                    XNameBin = list_to_binary(XNameList1),
+                    XNameBin = unicode:characters_to_binary(XNameList1),
                     XName = rabbit_misc:r(Vhost, exchange, XNameBin),
                     {ok, X} = rabbit_exchange:lookup(XName),
                     check_internal_exchange(X),
@@ -1933,7 +1941,7 @@ ensure_target(#'v1_0.target'{address = Address,
                     RoutingKey = case RK of
                                      undefined -> undefined;
                                      []        -> undefined;
-                                     _         -> list_to_binary(RK)
+                                     _         -> unicode:characters_to_binary(RK)
                                  end,
                     {ok, Exchange, RoutingKey, QNameBin};
                 {error, _} = E ->
@@ -2117,8 +2125,8 @@ ensure_source(#'v1_0.source'{address = Address,
                             true = string:equal(QNameList, QNameBin),
                             {ok, QNameBin};
                         {XNameList, RoutingKeyList} ->
-                            RoutingKey = list_to_binary(RoutingKeyList),
-                            XNameBin = list_to_binary(XNameList),
+                            RoutingKey = unicode:characters_to_binary(RoutingKeyList),
+                            XNameBin = unicode:characters_to_binary(XNameList),
                             XName = rabbit_misc:r(Vhost, exchange, XNameBin),
                             QName = rabbit_misc:r(Vhost, queue, QNameBin),
                             Binding = #binding{source = XName,
@@ -2322,19 +2330,19 @@ ensure_terminus(target, {queue, undefined}, _, _, _) ->
     %% Default exchange exists.
     undefined;
 ensure_terminus(_, {queue, QNameList}, Vhost, User, Durability) ->
-    declare_queue(list_to_binary(QNameList), Vhost, User, Durability);
+    declare_queue(unicode:characters_to_binary(QNameList), Vhost, User, Durability);
 ensure_terminus(_, {amqqueue, QNameList}, Vhost, _, _) ->
     %% Target "/amq/queue/" is handled specially due to AMQP legacy:
     %% "Queue names starting with "amq." are reserved for pre-declared and
     %% standardised queues. The client MAY declare a queue starting with "amq."
     %% if the passive option is set, or the queue already exists."
-    QNameBin = list_to_binary(QNameList),
+    QNameBin = unicode:characters_to_binary(QNameList),
     ok = exit_if_absent(queue, Vhost, QNameBin),
     QNameBin.
 
-exit_if_absent(Type, Vhost, Name) ->
-    ResourceName = rabbit_misc:r(Vhost, Type, rabbit_data_coercion:to_binary(Name)),
-    Mod = case Type of
+exit_if_absent(Kind, Vhost, Name) ->
+    ResourceName = rabbit_misc:r(Vhost, Kind, unicode:characters_to_binary(Name)),
+    Mod = case Kind of
               exchange -> rabbit_exchange;
               queue -> rabbit_amqqueue
           end,

deps/rabbit/src/rabbit_amqp_session.erl

@@ -2420,6 +2420,7 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
                                                 Args0),
     StrippedQueueNameBin = strip_cr_lf(QueueNameBin),
     Durable = DurableDeclare andalso not ExclusiveDeclare,
+    Kind = queue,
     ActualNameBin = case StrippedQueueNameBin of
                         <<>> ->
                             case rabbit_amqqueue:is_server_named_allowed(Args) of
@@ -2431,9 +2432,9 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
                                       "Cannot declare a server-named queue for type ~tp",
                                       [rabbit_amqqueue:get_queue_type(Args)])
                             end;
-                        Other -> check_name('queue', Other)
+                        Other -> check_name(Kind, Other)
                     end,
-    QueueName = rabbit_misc:r(VHostPath, queue, ActualNameBin),
+    QueueName = rabbit_misc:r(VHostPath, Kind, ActualNameBin),
     check_configure_permitted(QueueName, User, AuthzContext),
     rabbit_core_metrics:queue_declared(QueueName),
     case rabbit_amqqueue:with(

deps/rabbit/src/rabbit_channel.erl

@@ -167,8 +167,7 @@ init_per_testcase(T, Config)
        T =:= roundtrip_with_drain_quorum_queue orelse
        T =:= timed_get_quorum_queue orelse
        T =:= available_messages_quorum_queue ->
-    case rabbit_ct_broker_helpers:rpc(
-           Config, rabbit_feature_flags, is_enabled, [credit_api_v2]) of
+    case rpc(Config, rabbit_feature_flags, is_enabled, [credit_api_v2]) of
         true ->
             rabbit_ct_helpers:testcase_started(Config, T);
         false ->
@@ -2322,9 +2321,8 @@ async_notify(SenderSettleMode, QType, Config) ->
             %% If it is a stream we need to wait until there is a local member
             %% on the node we want to subscibe from before proceeding.
             rabbit_ct_helpers:await_condition(
-              fun() -> rabbit_ct_broker_helpers:rpc(
-                         Config, 0, ?MODULE, has_local_member,
-                         [rabbit_misc:r(<<"/">>, queue, QName)])
+              fun() -> rpc(Config, 0, ?MODULE, has_local_member,
+                           [rabbit_misc:r(<<"/">>, queue, QName)])
               end, 30_000);
         _ ->
             ok

deps/rabbit/test/amqp_client_SUITE.erl

@@ -21,7 +21,7 @@ parse_endpoint(Destination, AllowAnonymousQueue) when is_binary(Destination) ->
     parse_endpoint(unicode:characters_to_list(Destination),
                                               AllowAnonymousQueue);
 parse_endpoint(Destination, AllowAnonymousQueue) when is_list(Destination) ->
-    case re:split(Destination, "/", [{return, list}]) of
+    case re:split(Destination, "/", [unicode, {return, list}]) of
         [Name] ->
             {ok, {queue, unescape(Name)}};
         ["", Type | Rest]