Simplify splitting large messages

Author: ansd

deps/amqp10_client/src/amqp10_client_internal.hrl

@@ -7,7 +7,6 @@
 
 -define(AMQP_PROTOCOL_HEADER, <<"AMQP", 0, 1, 0, 0>>).
 -define(SASL_PROTOCOL_HEADER, <<"AMQP", 3, 1, 0, 0>>).
--define(FRAME_HEADER_SIZE, 8).
 
 -define(TIMEOUT, 5000).
 

deps/amqp10_client/src/amqp10_client_session.erl

@@ -636,7 +636,7 @@ send_transfer(Transfer0, Sections0, FooterOpt, MaxMessageSize,
        true ->
            % TODO: this does not take the extended header into account
            % see: 2.3
-           MaxPayloadSize = OutMaxFrameSize - TransferSize - ?FRAME_HEADER_SIZE,
+           MaxPayloadSize = OutMaxFrameSize - ?FRAME_HEADER_SIZE - TransferSize,
            Frames = build_frames(Channel, Transfer, SectionsBin, MaxPayloadSize, []),
            ok = socket_send(Socket, Frames),
            {ok, length(Frames)}

deps/amqp10_common/include/amqp10_types.hrl

@@ -2,6 +2,10 @@
 
 % [1.6.5]
 -type uint() :: 0..?UINT_MAX.
+
+% [2.3.1]
+-define(FRAME_HEADER_SIZE, 8).
+
 % [2.8.4]
 -type link_handle() :: uint().
 % [2.8.8]

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -827,14 +827,9 @@ send_to_new_session(
                                   container_id = ContainerId,
                                   name = ConnName},
       writer = WriterPid} = State) ->
-    %% Subtract fixed frame header size.
-    OutgoingMaxFrameSize = case MaxFrame of
-                               unlimited -> unlimited;
-                               _ -> MaxFrame - 8
-                           end,
     ChildArgs = [WriterPid,
                  ChannelNum,
-                 OutgoingMaxFrameSize,
+                 MaxFrame,
                  User,
                  Vhost,
                  ContainerId,

deps/rabbit/src/rabbit_amqp_session.erl

@@ -3163,19 +3163,20 @@ transfer_frames(Transfer, Sections, unlimited) ->
     [[Transfer, Sections]];
 transfer_frames(Transfer, Sections, MaxFrameSize) ->
     PerformativeSize = iolist_size(amqp10_framing:encode_bin(Transfer)),
-    encode_frames(Transfer, Sections, MaxFrameSize - PerformativeSize, []).
+    MaxPayloadSize = MaxFrameSize - ?FRAME_HEADER_SIZE - PerformativeSize,
+    split_msg(Transfer, Sections, MaxPayloadSize, []).
 
-encode_frames(_T, _Msg, MaxPayloadSize, _Transfers) when MaxPayloadSize =< 0 ->
+split_msg(_T, _Msg, MaxPayloadSize, _Transfers) when MaxPayloadSize =< 0 ->
     protocol_error(?V_1_0_AMQP_ERROR_FRAME_SIZE_TOO_SMALL,
                    "Frame size is too small by ~b bytes",
                    [-MaxPayloadSize]);
-encode_frames(T, Msg, MaxPayloadSize, Transfers) ->
+split_msg(T, Msg, MaxPayloadSize, Transfers) ->
     case iolist_size(Msg) > MaxPayloadSize of
         true ->
             MsgBin = iolist_to_binary(Msg),
             {Chunk, Rest} = split_binary(MsgBin, MaxPayloadSize),
             T1 = T#'v1_0.transfer'{more = true},
-            encode_frames(T, Rest, MaxPayloadSize, [[T1, Chunk] | Transfers]);
+            split_msg(T, Rest, MaxPayloadSize, [[T1, Chunk] | Transfers]);
         false ->
             lists:reverse([[T, Msg] | Transfers])
     end.

deps/rabbit/test/amqp_dotnet_SUITE_data/fsharp-tests/Program.fs

@@ -279,7 +279,11 @@ module Test =
 
     let fragmentation uri =
         for frameSize, size in
-            [1024u, 1024
+            [1024u, 990
+             1024u, 1000
+             1024u, 1010
+             1024u, 1020
+             1024u, 1024
              1024u, 1100
              1024u, 2048
              2048u, 2048] do