Use amqp-value section instead of data section

Author: ansd

deps/amqp10_common/src/amqp10_framing.erl

@@ -120,10 +120,10 @@ decode({described, Descriptor, {map, Fields}}) ->
         Else ->
             fill_from_map(Else, Fields)
     end;
-decode({described, Descriptor, {binary, Field}}) ->
+decode({described, Descriptor, {binary, Field} = BinaryField}) ->
     case amqp10_framing0:record_for(Descriptor) of
         #'v1_0.amqp_value'{} ->
-            #'v1_0.amqp_value'{content = {binary, Field}};
+            #'v1_0.amqp_value'{content = BinaryField};
         #'v1_0.data'{} ->
             #'v1_0.data'{content = Field}
     end;

deps/rabbit/src/rabbit_amqp_management.erl

@@ -5,32 +5,22 @@
 
 -export([handle_request/4]).
 
-%% An HTTP message mapped to AMQP using projected mode
-%% [HTTP over AMQP Working Draft 06 §4.1]
--record(msg,
-        {
-         properties :: #'v1_0.properties'{},
-         application_properties :: list(),
-         data = [] :: [#'v1_0.data'{}]
-        }).
-
 -spec handle_request(binary(), rabbit_types:vhost(), rabbit_types:user(), pid()) -> iolist().
 handle_request(Request, Vhost, User, ConnectionPid) ->
     ReqSections = amqp10_framing:decode_bin(Request),
     ?DEBUG("~s Inbound request:~n  ~tp",
            [?MODULE, [amqp10_framing:pprint(Section) || Section <- ReqSections]]),
-    #msg{properties = #'v1_0.properties'{
-                         message_id = MessageId,
-                         to = {utf8, HttpRequestTarget},
-                         subject = {utf8, HttpMethod},
-                         %% see Link Pair CS 01 §2.1
-                         %% https://docs.oasis-open.org/amqp/linkpair/v1.0/cs01/linkpair-v1.0-cs01.html#_Toc51331305
-                         reply_to = {utf8, <<"$me">>}},
-         application_properties = _OtherHttpHeaders,
-         data = ReqBody
-        } = decode_req(ReqSections, {undefined, undefined, []}),
+    {#'v1_0.properties'{
+        message_id = MessageId,
+        to = {utf8, HttpRequestTarget},
+        subject = {utf8, HttpMethod},
+        %% see Link Pair CS 01 §2.1
+        %% https://docs.oasis-open.org/amqp/linkpair/v1.0/cs01/linkpair-v1.0-cs01.html#_Toc51331305
+        reply_to = {utf8, <<"$me">>}},
+     ReqBin
+    } = decode_req(ReqSections, {undefined, undefined}),
     {PathSegments, QueryMap} = parse_uri(HttpRequestTarget),
-    ReqPayload = amqp10_framing:decode_bin(list_to_binary(ReqBody)),
+    ReqPayload = amqp10_framing:decode_bin(ReqBin),
     {RespProps0,
      RespAppProps0 = #'v1_0.application_properties'{content = C},
      RespPayload} = handle_http_req(HttpMethod,
@@ -47,7 +37,8 @@ handle_request(Request, Vhost, User, ConnectionPid) ->
                              correlation_id = MessageId},
     RespAppProps = RespAppProps0#'v1_0.application_properties'{
                                    content = [{{utf8, <<"http:response">>}, {utf8, <<"1.1">>}} | C]},
-    RespDataSect = #'v1_0.data'{content = iolist_to_binary(amqp10_framing:encode_bin(RespPayload))},
+    RespBody = iolist_to_binary(amqp10_framing:encode_bin(RespPayload)),
+    RespDataSect = #'v1_0.amqp_value'{content = {binary, RespBody}},
     RespSections = [RespProps, RespAppProps, RespDataSect],
     [amqp10_framing:encode_bin(Sect) || Sect <- RespSections].
 
@@ -79,7 +70,7 @@ handle_http_req(<<"PUT">>,
     {new, _Q} = rabbit_queue_type:declare(Q0, node()),
     Props = #'v1_0.properties'{subject = {utf8, <<"201">>}},
     AppProps = #'v1_0.application_properties'{content = []},
-    RespPayload = {map, []},
+    RespPayload = undefined,
     {Props, AppProps, RespPayload};
 
 handle_http_req(<<"PUT">>,
@@ -107,14 +98,13 @@ handle_http_req(<<"PUT">>,
                                  Username),
     Props = #'v1_0.properties'{subject = {utf8, <<"201">>} },
     AppProps = #'v1_0.application_properties'{content = []},
-
-    RespPayload = {map, []},
+    RespPayload = undefined,
     {Props, AppProps, RespPayload};
 
 handle_http_req(<<"DELETE">>,
                 [<<"queues">>, QNameBinQ, <<"messages">>],
                 _Query,
-                [],
+                [undefined],
                 Vhost,
                 _User,
                 ConnPid) ->
@@ -133,7 +123,7 @@ handle_http_req(<<"DELETE">>,
 handle_http_req(<<"DELETE">>,
                 [<<"queues">>, QNameBinQ],
                 _Query,
-                [],
+                [undefined],
                 Vhost,
                 #user{username = Username},
                 ConnPid) ->
@@ -148,7 +138,7 @@ handle_http_req(<<"DELETE">>,
 handle_http_req(<<"DELETE">>,
                 [<<"exchanges">>, XNameBinQ],
                 _Query,
-                [],
+                [undefined],
                 Vhost,
                 #user{username = Username},
                 _ConnPid) ->
@@ -164,7 +154,7 @@ handle_http_req(<<"DELETE">>,
          end,
     Props = #'v1_0.properties'{subject = {utf8, <<"204">>}},
     AppProps = #'v1_0.application_properties'{content = []},
-    RespPayload = {map, []},
+    RespPayload = undefined,
     {Props, AppProps, RespPayload};
 
 handle_http_req(<<"POST">>,
@@ -195,13 +185,13 @@ handle_http_req(<<"POST">>,
     Location = compose_binding_uri(SrcXNameBin, DstKind, DstNameBin, BindingKey, Args),
     AppProps = #'v1_0.application_properties'{
                   content = [{{utf8, <<"location">>}, {utf8, Location}}]},
-    RespPayload = {map, []},
+    RespPayload = undefined,
     {Props, AppProps, RespPayload};
 
 handle_http_req(<<"DELETE">>,
                 [<<"bindings">>, BindingSegment],
                 _Query,
-                [],
+                [undefined],
                 Vhost,
                 #user{username = Username},
                 _ConnPid) ->
@@ -217,14 +207,14 @@ handle_http_req(<<"DELETE">>,
     end,
     Props = #'v1_0.properties'{subject = {utf8, <<"204">>}},
     AppProps = #'v1_0.application_properties'{content = []},
-    RespPayload = {map, []},
+    RespPayload = undefined,
     {Props, AppProps, RespPayload};
 
 handle_http_req(<<"GET">>,
                 [<<"bindings">>],
                 QueryMap = #{<<"src">> := SrcXNameBin,
                              <<"key">> := Key},
-                [],
+                [undefined],
                 Vhost,
                 _User,
                 _ConnPid) ->
@@ -326,17 +316,13 @@ encode_bindings(Bindings) ->
            end, Bindings),
     {list, Bs}.
 
-decode_req([], {Props, AppProps, DataRev}) ->
-    #msg{properties = Props,
-         application_properties = AppProps,
-         data = lists:reverse(DataRev)};
+decode_req([], Acc) ->
+    Acc;
 decode_req([#'v1_0.properties'{} = P | Rem], Acc) ->
     decode_req(Rem, setelement(1, Acc, P));
-decode_req([#'v1_0.application_properties'{content = Content} | Rem], Acc) ->
-    decode_req(Rem, setelement(2, Content, Acc));
-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([#'v1_0.amqp_value'{content = {binary, B}} | Rem], Acc) ->
+    decode_req(Rem, setelement(2, Acc, B));
+decode_req([_IgnoreSection | Rem], Acc) ->
     decode_req(Rem, Acc).
 
 parse_uri(Uri) ->

deps/rabbitmq_amqp_client/app.bzl

@@ -13,6 +13,7 @@ def all_beam_files(name = "all_beam_files"):
         app_name = "rabbitmq_amqp_client",
         dest = "ebin",
         erlc_opts = "//:erlc_opts",
+        deps = ["//deps/amqp10_common:erlang_app"],
     )
 
 def all_srcs(name = "all_srcs"):
@@ -48,6 +49,7 @@ def all_test_beam_files(name = "all_test_beam_files"):
         app_name = "rabbitmq_amqp_client",
         dest = "test",
         erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp10_common:erlang_app"],
     )
     filegroup(
         name = "test_beam_files",

deps/rabbitmq_amqp_client/src/rabbitmq_amqp_client.erl

@@ -8,6 +8,8 @@
 
 -feature(maybe_expr, enable).
 
+-include_lib("amqp10_common/include/amqp10_framing.hrl").
+
 -export[attach_management_link_pair_sync/2,
         detach_management_link_pair_sync/1,
         declare_queue/2,
@@ -140,9 +142,7 @@ declare_queue(LinkPair, QueueProperties) ->
                                           end, [], V),
                                {N, [{{utf8, <<"arguments">>}, {map, KVList}} | L0]}
                        end, {undefined, []}, QueueProperties),
-    Body1 = {map, Body0},
-    Body = iolist_to_binary(amqp10_framing:encode_bin(Body1)),
-
+    Body = {map, Body0},
     QNameQuoted = uri_string:quote(QName),
     Props = #{subject => <<"PUT">>,
               to => <<"/queues/", QNameQuoted/binary>>},
@@ -151,9 +151,10 @@ declare_queue(LinkPair, QueueProperties) ->
         {ok, Resp} ->
             case amqp10_msg:properties(Resp) of
                 #{subject := <<"201">>} ->
-                    RespBody = amqp10_msg:body_bin(Resp),
-                    [{map, KVList}] = amqp10_framing:decode_bin(RespBody),
-                    {ok, proplists:to_map(KVList)};
+                    #'v1_0.amqp_value'{content = {binary, RespBin}} = amqp10_msg:body(Resp),
+                    [undefined] = amqp10_framing:decode_bin(RespBin),
+                    % {ok, proplists:to_map(KVList)};
+                    {ok, #{}};
                 Other ->
                     {error, Other}
             end;
@@ -179,14 +180,12 @@ bind(DestinationKind, LinkPair, Destination, Source, BindingKey, BindingArgument
                      when is_binary(Key) ->
                        [{{utf8, Key}, TaggedVal} | L]
                end, [], BindingArguments),
-    Body0 = {map, [
-                   {{utf8, <<"source">>}, {utf8, Source}},
-                   {{utf8, DestinationKind}, {utf8, Destination}},
-                   {{utf8, <<"binding_key">>}, {utf8, BindingKey}},
-                   {{utf8, <<"arguments">>}, {map, KVList}}
-                  ]},
-    Body = iolist_to_binary(amqp10_framing:encode_bin(Body0)),
-
+    Body = {map, [
+                  {{utf8, <<"source">>}, {utf8, Source}},
+                  {{utf8, DestinationKind}, {utf8, Destination}},
+                  {{utf8, <<"binding_key">>}, {utf8, BindingKey}},
+                  {{utf8, <<"arguments">>}, {map, KVList}}
+                 ]},
     Props = #{subject => <<"POST">>,
               to => <<"/bindings">>},
 
@@ -235,12 +234,12 @@ unbind(DestinationChar, LinkPair, Destination, Source, BindingKey, BindingArgume
     Props = #{subject => <<"GET">>,
               to => Uri0},
 
-    case request(LinkPair, Props, <<>>) of
+    case request(LinkPair, Props, undefined) of
         {ok, Resp} ->
             case amqp10_msg:properties(Resp) of
                 #{subject := <<"200">>} ->
-                    RespBody = amqp10_msg:body_bin(Resp),
-                    [{list, Bindings}] = amqp10_framing:decode_bin(RespBody),
+                    #'v1_0.amqp_value'{content = {binary, RespBin}} = amqp10_msg:body(Resp),
+                    [{list, Bindings}] = amqp10_framing:decode_bin(RespBin),
                     case search_binding_uri(BindingArguments, Bindings) of
                         {ok, Uri} ->
                             delete_binding(LinkPair, Uri);
@@ -278,7 +277,7 @@ search_binding_uri(BindingArguments, [{map, Binding} | Bindings]) ->
 delete_binding(LinkPair, BindingUri) ->
     Props = #{subject => <<"DELETE">>,
               to => BindingUri},
-    case request(LinkPair, Props, <<>>) of
+    case request(LinkPair, Props, undefined) of
         {ok, Resp} ->
             case amqp10_msg:properties(Resp) of
                 #{subject := <<"204">>} ->
@@ -307,15 +306,15 @@ purge_or_delete_queue(LinkPair, QueueName, PathSuffix) ->
     HttpRequestTarget = <<"/queues/", QNameQuoted/binary, PathSuffix/binary>>,
     Props = #{subject => <<"DELETE">>,
               to => HttpRequestTarget},
-    case request(LinkPair, Props, <<>>) of
+    case request(LinkPair, Props, undefined) of
         {ok, Resp} ->
             case amqp10_msg:properties(Resp) of
                 #{subject := <<"200">>} ->
-                    RespBody = amqp10_msg:body_bin(Resp),
+                    #'v1_0.amqp_value'{content = {binary, RespBin}} = amqp10_msg:body(Resp),
                     [{map, [
                             {{utf8, <<"message_count">>}, {ulong, Count}}
                            ]
-                     }] = amqp10_framing:decode_bin(RespBody),
+                     }] = amqp10_framing:decode_bin(RespBin),
                     {ok, #{message_count => Count}};
                 _ ->
                     {error, Resp}
@@ -325,7 +324,7 @@ purge_or_delete_queue(LinkPair, QueueName, PathSuffix) ->
     end.
 
 -spec declare_exchange(link_pair(), exchange_properties()) ->
-    {ok, map()} | {error, term()}.
+    ok | {error, term()}.
 declare_exchange(LinkPair, ExchangeProperties) ->
     {XName, Body0} = maps:fold(
                        fun(name, V, {undefined, L}) when is_binary(V) ->
@@ -346,8 +345,7 @@ declare_exchange(LinkPair, ExchangeProperties) ->
                                           end, [], V),
                                {N, [{{utf8, <<"arguments">>}, {map, KVList}} | L0]}
                        end, {undefined, []}, ExchangeProperties),
-    Body1 = {map, Body0},
-    Body = iolist_to_binary(amqp10_framing:encode_bin(Body1)),
+    Body = {map, Body0},
 
     XNameQuoted = uri_string:quote(XName),
     Props = #{subject => <<"PUT">>,
@@ -357,9 +355,9 @@ declare_exchange(LinkPair, ExchangeProperties) ->
         {ok, Resp} ->
             case amqp10_msg:properties(Resp) of
                 #{subject := <<"201">>} ->
-                    RespBody = amqp10_msg:body_bin(Resp),
-                    [{map, KVList}] = amqp10_framing:decode_bin(RespBody),
-                    {ok, proplists:to_map(KVList)};
+                    #'v1_0.amqp_value'{content = {binary, RespBin}} = amqp10_msg:body(Resp),
+                    [undefined] = amqp10_framing:decode_bin(RespBin),
+                    ok;
                 _ ->
                     {error, Resp}
             end;
@@ -373,7 +371,7 @@ delete_exchange(LinkPair, ExchangeName) ->
     XNameQuoted = uri_string:quote(ExchangeName),
     Props = #{subject => <<"DELETE">>,
               to => <<"/exchanges/", XNameQuoted/binary>>},
-    case request(LinkPair, Props, <<>>) of
+    case request(LinkPair, Props, undefined) of
         {ok, Resp} ->
             case amqp10_msg:properties(Resp) of
                 #{subject := <<"204">>} ->
@@ -385,14 +383,15 @@ delete_exchange(LinkPair, ExchangeName) ->
             Err
     end.
 
--spec request(link_pair(), amqp10_msg:amqp10_properties(), binary()) ->
+-spec request(link_pair(), amqp10_msg:amqp10_properties(), undefined | amqp10_prim()) ->
     {ok, Response :: amqp10_msg:amqp10_msg()} | {error, term()}.
 request(#link_pair{outgoing_link = OutgoingLink,
                    incoming_link = IncomingLink}, Properties, Body) ->
     MessageId = message_id(),
     Properties1 = Properties#{message_id => {binary, MessageId},
                               reply_to => <<"$me">>},
-    Request = amqp10_msg:new(<<>>, Body, true),
+    BodyBin = iolist_to_binary(amqp10_framing:encode_bin(Body)),
+    Request = amqp10_msg:new(<<>>, #'v1_0.amqp_value'{content = {binary, BodyBin}}, true),
     Request1 =  amqp10_msg:set_properties(Properties1, Request),
     ok = amqp10_client:flow_link_credit(IncomingLink, 1, never),
     case amqp10_client:send_msg(OutgoingLink, Request1) of

deps/rabbitmq_amqp_client/test/management_SUITE.erl

@@ -117,7 +117,7 @@ all_management_operations(Config) ->
           auto_delete => false,
           internal => false,
           arguments => #{}},
-    {ok, #{}} = rabbitmq_amqp_client:declare_exchange(LinkPair, X),
+    ?assertEqual(ok, rabbitmq_amqp_client:declare_exchange(LinkPair, X)),
 
     TargetAddr3 = <<"/exchange/", XName/binary>>,
     SourceExchange = <<"amq.direct">>,