amqp10_client

%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
%%
-module(amqp10_client_session).

-behaviour(gen_statem).

-include("amqp10_client.hrl").
-include_lib("amqp10_common/include/amqp10_framing.hrl").
-include_lib("amqp10_common/include/amqp10_types.hrl").
-include_lib("kernel/include/logger.hrl").

%% Public API.
-export(['begin'/1,
         begin_sync/1,
         begin_sync/2,
         'end'/1,
         attach/2,
         detach/2,
         transfer/3,
         disposition/5,
         flow_link/4
        ]).

%% Manual session flow control is currently only used in tests.
-export([flow/3]).

%% Private API
-export([start_link/4,
         socket_ready/2
        ]).

%% gen_statem callbacks
-export([
         init/1,
         terminate/3,
         code_change/4,
         callback_mode/0,
         format_status/1
        ]).

%% gen_statem state callbacks
%% see figure 2.30
-export([
         unmapped/3,
         begin_sent/3,
         mapped/3,
         end_sent/3
        ]).

-import(serial_number,
        [add/2,
         diff/2]).

%% By default, we want to keep the server's remote-incoming-window large at all times.
-define(DEFAULT_MAX_INCOMING_WINDOW, 100_000).
-define(UINT_OUTGOING_WINDOW, {uint, ?UINT_MAX}).
-define(INITIAL_OUTGOING_DELIVERY_ID, ?UINT_MAX).
%% "The next-outgoing-id MAY be initialized to an arbitrary value" [2.5.6]
-define(INITIAL_OUTGOING_TRANSFER_ID, ?UINT_MAX - 1).
%% "Note that, despite its name, the delivery-count is not a count but a
%% sequence number initialized at an arbitrary point by the sender." [2.6.7]
-define(INITIAL_DELIVERY_COUNT, ?UINT_MAX - 2).

-type link_name() :: binary().
%% https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-address-string
%% or
%% https://docs.oasis-open.org/amqp/anonterm/v1.0/anonterm-v1.0.html
-type terminus_address() :: binary() | null.
-type link_role() :: sender | receiver.
-type link_target() :: {pid, pid()} | terminus_address() | undefined.
%% "The locally chosen handle is referred to as the output handle." [2.6.2]
-type output_handle() :: link_handle().
%% "The remotely chosen handle is referred to as the input handle." [2.6.2]
-type input_handle() :: link_handle().

-type terminus_durability() :: none | configuration | unsettled_state.

-type target_def() :: #{address => terminus_address(),
                        durable => terminus_durability()}.
-type source_def() :: #{address => terminus_address(),
                        durable => terminus_durability()}.

-type attach_role() :: {sender, target_def()} | {receiver, source_def(), pid()}.

% http://www.amqp.org/specification/1.0/filters
-type filter() :: #{binary() => binary() | map() | list(binary())}.
-type max_message_size() :: undefined | non_neg_integer().
-type footer_opt() :: crc32 | adler32.

-type attach_args() :: #{name => binary(),
                         role => attach_role(),
                         snd_settle_mode => snd_settle_mode(),
                         rcv_settle_mode => rcv_settle_mode(),
                         filter => filter(),
                         properties => amqp10_client_types:properties(),
                         max_message_size => max_message_size(),
                         handle => output_handle(),
                         footer_opt => footer_opt()
                        }.

-type transfer_error() :: {error,
                           insufficient_credit |
                           remote_incoming_window_exceeded |
                           message_size_exceeded |
                           link_not_found |
                           half_attached}.

-type link_ref() :: #link_ref{}.

-export_type([snd_settle_mode/0,
              rcv_settle_mode/0,
              terminus_durability/0,
              attach_args/0,
              attach_role/0,
              terminus_address/0,
              target_def/0,
              source_def/0,
              filter/0,
              max_message_size/0,
              transfer_error/0]).

-record(link,
        {name :: link_name(),
         ref :: link_ref(),
         state = detached :: detached | attach_sent | attached | detach_sent,
         notify :: pid(),
         output_handle :: output_handle(),
         input_handle :: input_handle() | undefined,
         role :: link_role(),
         target :: link_target(),
         max_message_size :: non_neg_integer() | Unlimited :: undefined,
         delivery_count :: sequence_no() | undefined,
         link_credit = 0 :: non_neg_integer(),
         available = 0 :: non_neg_integer(),
         drain = false :: boolean(),
         partial_transfers :: undefined | {#'v1_0.transfer'{}, [binary()]},
         auto_flow :: never | {RenewWhenBelow :: pos_integer(),
                               Credit :: pos_integer()},
         incoming_unsettled = #{} :: #{delivery_number() => ok},
         footer_opt :: footer_opt() | undefined
        }).

-record(state,
        {channel :: pos_integer(),
         remote_channel :: pos_integer() | undefined,

         %% session flow control, see section 2.5.6
         next_incoming_id :: transfer_number() | undefined,
         %% Can become negative if the peer overshoots our window.
         incoming_window :: integer(),
         auto_flow :: never | {RenewWhenBelow :: pos_integer(),
                               NewWindowSize :: pos_integer()},
         next_outgoing_id = ?INITIAL_OUTGOING_TRANSFER_ID :: transfer_number(),
         remote_incoming_window = 0 :: non_neg_integer(),
         remote_outgoing_window = 0 :: non_neg_integer(),

         reader :: pid(),
         socket :: amqp10_client_socket:socket() | undefined,
         links = #{} :: #{output_handle() => #link{}},
         link_index = #{} :: #{{link_role(), link_name()} => output_handle()},
         link_handle_index = #{} :: #{input_handle() => output_handle()},
         next_link_handle = 0 :: output_handle(),
         early_attach_requests :: [term()],
         connection_config :: amqp10_client_connection:connection_config(),
         outgoing_delivery_id = ?INITIAL_OUTGOING_DELIVERY_ID :: delivery_number(),
         outgoing_unsettled = #{} :: #{delivery_number() => {amqp10_msg:delivery_tag(), Notify :: pid()}},
         notify :: pid()
        }).

%% -------------------------------------------------------------------
%% Public API.
%% -------------------------------------------------------------------

-spec 'begin'(pid()) -> supervisor:startchild_ret().
'begin'(Connection) ->
    %% The connection process is responsible for allocating a channel
    %% number and contact the sessions supervisor to start a new session
    %% process.
    amqp10_client_connection:begin_session(Connection).

-spec begin_sync(pid()) -> supervisor:startchild_ret().
begin_sync(Connection) ->
    begin_sync(Connection, ?TIMEOUT).

-spec begin_sync(pid(), non_neg_integer()) ->
    supervisor:startchild_ret() | session_timeout.
begin_sync(Connection, Timeout) ->
    {ok, Session} = amqp10_client_connection:begin_session(Connection),
    receive
        {session_begin, Session} -> {ok, Session}
    after Timeout -> session_timeout
    end.

-spec 'end'(pid()) -> ok.
'end'(Pid) ->
    gen_statem:cast(Pid, 'end').

-spec attach(pid(), attach_args()) -> {ok, link_ref()}.
attach(Session, Args) ->
    gen_statem:call(Session, {attach, Args}, ?TIMEOUT).

-spec detach(pid(), output_handle()) -> ok | {error, link_not_found | half_attached}.
detach(Session, Handle) ->
    gen_statem:call(Session, {detach, Handle}, ?TIMEOUT).

-spec transfer(pid(), amqp10_msg:amqp10_msg(), timeout()) ->
    ok | transfer_error().
transfer(Session, Amqp10Msg, Timeout) ->
    [Transfer | Sections] = amqp10_msg:to_amqp_records(Amqp10Msg),
    gen_statem:call(Session, {transfer, Transfer, Sections}, Timeout).

-spec flow(pid(), non_neg_integer(), never | pos_integer()) -> ok.
flow(Session, IncomingWindow, RenewWhenBelow) when
      %% Check that the RenewWhenBelow value make sense.
      RenewWhenBelow =:= never orelse
      is_integer(RenewWhenBelow) andalso
      RenewWhenBelow > 0 andalso
      RenewWhenBelow =< IncomingWindow ->
    gen_statem:cast(Session, {flow_session, IncomingWindow, RenewWhenBelow}).

-spec flow_link(pid(), link_handle(), #'v1_0.flow'{}, never | pos_integer()) -> ok.
flow_link(Session, Handle, Flow, RenewWhenBelow) ->
    gen_statem:cast(Session, {flow_link, Handle, Flow, RenewWhenBelow}).

%% Sending a disposition on a sender link (with receiver-settle-mode = second)
%% is currently unsupported.
-spec disposition(link_ref(), delivery_number(), delivery_number(), boolean(),
                  amqp10_client_types:delivery_state()) -> ok.
disposition(#link_ref{role = receiver,
                      session = Session,
                      link_handle = Handle},
            First, Last, Settled, DeliveryState) ->
    gen_statem:call(Session, {disposition, Handle, First, Last, Settled,
                              DeliveryState}, ?TIMEOUT).


%% -------------------------------------------------------------------
%% Private API.
%% -------------------------------------------------------------------

start_link(From, Channel, Reader, ConnConfig) ->
    gen_statem:start_link(?MODULE, [From, Channel, Reader, ConnConfig], []).

-spec socket_ready(pid(), amqp10_client_socket:socket()) -> ok.
socket_ready(Pid, Socket) ->
    gen_statem:cast(Pid, {socket_ready, Socket}).

%% -------------------------------------------------------------------
%% gen_statem callbacks.
%% -------------------------------------------------------------------

callback_mode() -> [state_functions].

init([FromPid, Channel, Reader, ConnConfig]) ->
    process_flag(trap_exit, true),
    amqp10_client_frame_reader:register_session(Reader, self(), Channel),
    State = #state{notify = FromPid,
                   channel = Channel,
                   reader = Reader,
                   connection_config = ConnConfig,
                   incoming_window = ?DEFAULT_MAX_INCOMING_WINDOW,
                   auto_flow = {?DEFAULT_MAX_INCOMING_WINDOW div 2,
                                ?DEFAULT_MAX_INCOMING_WINDOW},
                   early_attach_requests = []},
    {ok, unmapped, State}.

unmapped(cast, {socket_ready, Socket}, State) ->
    State1 = State#state{socket = Socket},
    ok = send_begin(State1),
    {next_state, begin_sent, State1};
unmapped({call, From}, {attach, Attach},
                      #state{early_attach_requests = EARs} = State) ->
    {keep_state,
     State#state{early_attach_requests = [{From, Attach} | EARs]}}.

begin_sent(cast, #'v1_0.begin'{remote_channel = {ushort, RemoteChannel},
                               next_outgoing_id = {uint, NOI},
                               incoming_window = {uint, InWindow},
                               outgoing_window = {uint, OutWindow}} = Begin,
           #state{early_attach_requests = EARs} = State) ->

    State1 = State#state{remote_channel = RemoteChannel},
    State2 = lists:foldr(fun({From, Attach}, S) ->
                                 {S2, H} = send_attach(fun send/2, Attach, From, S),
                                 gen_statem:reply(From, {ok, H}),
                                 S2
                         end, State1, EARs),

    ok = notify_session_begun(Begin, State2),

    {next_state, mapped, State2#state{early_attach_requests = [],
                                      next_incoming_id = NOI,
                                      remote_incoming_window = InWindow,
                                      remote_outgoing_window = OutWindow}};
begin_sent({call, From}, {attach, Attach},
           #state{early_attach_requests = EARs} = State) ->
    {keep_state,
     State#state{early_attach_requests = [{From, Attach} | EARs]}}.

mapped(cast, 'end', State) ->
    %% We send the first end frame and wait for the reply.
    send_end(State),
    {next_state, end_sent, State};
mapped(cast, {flow_link, OutHandle, Flow0, RenewWhenBelow}, State0) ->
    State = send_flow_link(OutHandle, Flow0, RenewWhenBelow, State0),
    {keep_state, State};
mapped(cast, {flow_session, IncomingWindow, RenewWhenBelow}, State0) ->
    AutoFlow = case RenewWhenBelow of
                   never -> never;
                   _ -> {RenewWhenBelow, IncomingWindow}
               end,
    State = State0#state{incoming_window = IncomingWindow,
                         auto_flow = AutoFlow},
    send_flow_session(State),
    {keep_state, State};
mapped(cast, #'v1_0.end'{} = End, State) ->
    %% We receive the first end frame, reply and terminate.
    _ = send_end(State),
    % TODO: send notifications for links?
    ok = notify_session_ended(End, State),
    {stop, normal, State};
mapped(cast, #'v1_0.attach'{name = {utf8, Name},
                            initial_delivery_count = IDC,
                            handle = {uint, InHandle},
                            role = PeerRoleBool,
                            max_message_size = MaybeMaxMessageSize} = Attach,
       #state{links = Links, link_index = LinkIndex,
              link_handle_index = LHI} = State0) ->

    OurRoleBool = not PeerRoleBool,
    OurRole = boolean_to_role(OurRoleBool),
    LinkIndexKey = {OurRole, Name},
    #{LinkIndexKey := OutHandle} = LinkIndex,
    #{OutHandle := Link0} = Links,
    ok = notify_link_attached(Link0, Attach, State0),

    {DeliveryCount, MaxMessageSize} =
    case Link0 of
        #link{role = sender = OurRole,
              delivery_count = DC} ->
            MSS = case MaybeMaxMessageSize of
                      {ulong, S} when S > 0 -> S;
                      _ -> undefined
                  end,
            {DC, MSS};
        #link{role = receiver = OurRole,
              max_message_size = MSS} ->
            {unpack(IDC), MSS}
    end,
    Link = Link0#link{state = attached,
                      input_handle = InHandle,
                      delivery_count = DeliveryCount,
                      max_message_size = MaxMessageSize},
    State = State0#state{links = Links#{OutHandle := Link},
                         link_index = maps:remove(LinkIndexKey, LinkIndex),
                         link_handle_index = LHI#{InHandle => OutHandle}},
    {keep_state, State};
mapped(cast, #'v1_0.detach'{handle = {uint, InHandle}} = Detach,
       #state{links = Links, link_handle_index = LHI} = State0) ->
    with_link(InHandle, State0,
              fun (#link{output_handle = OutHandle} = Link, State) ->
                      ok = notify_link_detached(Link, Detach, State),
                      {keep_state,
                       State#state{links = maps:remove(OutHandle, Links),
                                   link_handle_index = maps:remove(InHandle, LHI)}}
              end);
mapped(cast, #'v1_0.flow'{handle = undefined} = Flow, State0) ->
    State = handle_session_flow(Flow, State0),
    {keep_state, State};
mapped(cast, #'v1_0.flow'{handle = {uint, InHandle}} = Flow,
       #state{links = Links} = State0) ->

    State = handle_session_flow(Flow, State0),

    {ok, #link{output_handle = OutHandle} = Link0} =
        find_link_by_input_handle(InHandle, State),

     % TODO: handle `send_flow` return tag
    {ok, Link} = handle_link_flow(Flow, Link0),
    ok = maybe_notify_link_credit(Link0, Link),
    Links1 = Links#{OutHandle := Link},
    State1 = State#state{links = Links1},
    {keep_state, State1};
mapped(cast, {#'v1_0.transfer'{handle = {uint, InHandle},
                               more = true} = Transfer, Payload},
       #state{links = Links} = State0) ->

    {ok, #link{output_handle = OutHandle} = Link} =
        find_link_by_input_handle(InHandle, State0),

    Link1 = append_partial_transfer(Transfer, Payload, Link),

    State = book_partial_transfer_received(
              State0#state{links = Links#{OutHandle => Link1}}),
    {keep_state, State};
mapped(cast, {Transfer0 = #'v1_0.transfer'{handle = {uint, InHandle}},
              Payload0}, State0) ->
    {ok, #link{target = {pid, TargetPid},
               ref = LinkRef,
               incoming_unsettled = Unsettled,
               footer_opt = FooterOpt
              } = Link0} = find_link_by_input_handle(InHandle, State0),

    {Transfer = #'v1_0.transfer'{settled = Settled,
                                 delivery_id = {uint, DeliveryId}},
     Payload, Link1} = complete_partial_transfer(Transfer0, Payload0, Link0),

    Link2 = case Settled of
                true ->
                    Link1;
                _ ->
                    %% "If not set on the first (or only) transfer for a (multi-transfer) delivery,
                    %% then the settled flag MUST be interpreted as being false." [2.7.5]
                    Link1#link{incoming_unsettled = Unsettled#{DeliveryId => ok}}
            end,
    case decode_as_msg(Transfer, Payload, FooterOpt) of
        {ok, Msg} ->
            % link bookkeeping
            % notify when credit is exhausted (link_credit = 0)
            % detach the Link with a transfer-limit-exceeded error code if further
            % transfers are received
            case book_transfer_received(State0, Link2) of
                {ok, Link3, State1} ->
                    % deliver
                    TargetPid ! {amqp10_msg, LinkRef, Msg},
                    State = auto_flow(Link3, State1),
                    {keep_state, State};
                {credit_exhausted, Link3, State} ->
                    TargetPid ! {amqp10_msg, LinkRef, Msg},
                    notify_credit_exhausted(Link3),
                    {keep_state, State};
                {transfer_limit_exceeded, Link3, State} ->
                    ?LOG_WARNING("transfer_limit_exceeded for link ~tp", [Link3]),
                    Link = detach_with_error_cond(Link3,
                                                  State,
                                                  ?V_1_0_LINK_ERROR_TRANSFER_LIMIT_EXCEEDED,
                                                  undefined),
                    {keep_state, update_link(Link, State)}
            end;
        {checksum_error, Expected, Actual} ->
            Description = lists:flatten(
                            io_lib:format(
                              "~s checksum error: expected ~b, actual ~b",
                              [FooterOpt, Expected, Actual])),
            ?LOG_WARNING("deteaching link ~tp due to ~s", [Link2, Description]),
            Link = detach_with_error_cond(Link2,
                                          State0,
                                          ?V_1_0_AMQP_ERROR_DECODE_ERROR,
                                          {utf8, unicode:characters_to_binary(Description)}),
            {keep_state, update_link(Link, State0)}
    end;

% role=true indicates the disposition is from a `receiver`. i.e. from the
% clients point of view these are dispositions relating to `sender`links
mapped(cast, #'v1_0.disposition'{role = true,
                                 settled = true,
                                 first = {uint, First},
                                 last = Last0,
                                 state = DeliveryState},
       #state{outgoing_unsettled = Unsettled0} = State) ->
    Last = case Last0 of
               undefined -> First;
               {uint, L} -> L
           end,
    % TODO: no good if the range becomes very large!! refactor
    Unsettled = serial_number:foldl(
                  fun(Id, Acc0) ->
                          case maps:take(Id, Acc0) of
                              {{DeliveryTag, Pid}, Acc} ->
                                  %% TODO: currently all modified delivery states
                                  %% will be translated to the old, `modified` atom.
                                  %% At some point we should translate into the
                                  %% full {modified, bool, bool, map) tuple.
                                  S = translate_delivery_state(DeliveryState),
                                  ok = notify_disposition(Pid, {S, DeliveryTag}),
                                  Acc;
                              error ->
                                  Acc0
                          end
                  end, Unsettled0, First, Last),

    {keep_state, State#state{outgoing_unsettled = Unsettled}};
mapped(cast, Frame, State) ->
    ?LOG_WARNING("Unhandled session frame ~tp in state ~tp",
                 [Frame, State]),
    {keep_state, State};
mapped({call, From},
       {transfer, _Transfer, _Sections},
       #state{remote_incoming_window = Window})
  when Window =< 0 ->
    {keep_state_and_data, {reply, From, {error, remote_incoming_window_exceeded}}};
mapped({call, From = {Pid, _}},
       {transfer, #'v1_0.transfer'{handle = {uint, OutHandle},
                                   delivery_tag = {binary, DeliveryTag},
                                   settled = false} = Transfer0, Sections},
       #state{outgoing_delivery_id = DeliveryId, links = Links,
              outgoing_unsettled = Unsettled} = State) ->
    case Links of
        #{OutHandle := #link{input_handle = undefined}} ->
            {keep_state_and_data, {reply, From, {error, half_attached}}};
        #{OutHandle := #link{link_credit = LC}} when LC =< 0 ->
            {keep_state_and_data, {reply, From, {error, insufficient_credit}}};
        #{OutHandle := Link = #link{max_message_size = MaxMessageSize,
                                    footer_opt = FooterOpt}} ->
            Transfer = Transfer0#'v1_0.transfer'{delivery_id = uint(DeliveryId)},
            case send_transfer(Transfer, Sections, FooterOpt, MaxMessageSize, State) of
                {ok, NumFrames} ->
                    State1 = State#state{outgoing_unsettled = Unsettled#{DeliveryId => {DeliveryTag, Pid}}},
                    {keep_state, book_transfer_send(NumFrames, Link, State1), {reply, From, ok}};
                Error ->
                    {keep_state_and_data, {reply, From, Error}}
            end;
        _ ->
            {keep_state_and_data, {reply, From, {error, link_not_found}}}

    end;
mapped({call, From},
       {transfer, #'v1_0.transfer'{handle = {uint, OutHandle}} = Transfer0,
        Sections}, #state{outgoing_delivery_id = DeliveryId,
                       links = Links} = State) ->
    case Links of
        #{OutHandle := #link{input_handle = undefined}} ->
            {keep_state_and_data, {reply, From, {error, half_attached}}};
        #{OutHandle := #link{link_credit = LC}} when LC =< 0 ->
            {keep_state_and_data, {reply, From, {error, insufficient_credit}}};
        #{OutHandle := Link = #link{max_message_size = MaxMessageSize,
                                    footer_opt = FooterOpt}} ->
            Transfer = Transfer0#'v1_0.transfer'{delivery_id = uint(DeliveryId)},
            case send_transfer(Transfer, Sections, FooterOpt, MaxMessageSize, State) of
                {ok, NumFrames} ->
                    {keep_state, book_transfer_send(NumFrames, Link, State), {reply, From, ok}};
                Error ->
                    {keep_state_and_data, {reply, From, Error}}
            end;
        _ ->
            {keep_state_and_data, {reply, From, {error, link_not_found}}}
    end;

mapped({call, From},
       {disposition, OutputHandle, First, Last, Settled0, DeliveryState},
       #state{links = Links} = State0) ->
    #{OutputHandle := Link0 = #link{incoming_unsettled = Unsettled0}} = Links,
    Unsettled = serial_number:foldl(fun maps:remove/2, Unsettled0, First, Last),
    Link = Link0#link{incoming_unsettled = Unsettled},
    State1 = State0#state{links = Links#{OutputHandle := Link}},
    State = auto_flow(Link, State1),
    Disposition = #'v1_0.disposition'{
                     role = translate_role(receiver),
                     first = {uint, First},
                     last = {uint, Last},
                     settled = Settled0,
                     state = translate_delivery_state(DeliveryState)},
    Res = send(Disposition, State),
    {keep_state, State, {reply, From, Res}};

mapped({call, From}, {attach, Attach}, State) ->
    {State1, LinkRef} = send_attach(fun send/2, Attach, From, State),
    {keep_state, State1, {reply, From, {ok, LinkRef}}};

mapped({call, From}, Msg, State) ->
    {Reply, State1} = send_detach(fun send/2, Msg, From, State),
    {keep_state, State1, {reply, From, Reply}};

mapped(_EvtType, Msg, _State) ->
    ?LOG_WARNING("amqp10_session: unhandled msg in mapped state ~W",
                 [Msg, 10]),
    keep_state_and_data.

end_sent(_EvtType, #'v1_0.end'{} = End, State) ->
    ok = notify_session_ended(End, State),
    {stop, normal, State};
end_sent(_EvtType, _Frame, _State) ->
    % just drop frames here
    keep_state_and_data.

terminate(Reason, _StateName, #state{channel = Channel,
                                     remote_channel = RemoteChannel,
                                     reader = Reader}) ->
    case Reason of
        normal -> amqp10_client_frame_reader:unregister_session(
                    Reader, self(), Channel, RemoteChannel);
        _      -> ok
    end,
    ok.

code_change(_OldVsn, StateName, State, _Extra) ->
    {ok, StateName, State}.

%% -------------------------------------------------------------------
%% Internal functions.
%% -------------------------------------------------------------------

send_begin(#state{socket = Socket,
                  next_outgoing_id = NextOutId,
                  incoming_window = InWin} = State) ->
    Begin = #'v1_0.begin'{next_outgoing_id = uint(NextOutId),
                          incoming_window = uint(InWin),
                          outgoing_window = ?UINT_OUTGOING_WINDOW},
    Frame = encode_frame(Begin, State),
    socket_send(Socket, Frame).

send_end(State) ->
    send_end(State, undefined).

send_end(#state{socket = Socket} = State, Cond) ->
    Err = #'v1_0.error'{condition = Cond},
    End = #'v1_0.end'{error = Err},
    Frame = encode_frame(End, State),
    socket_send(Socket, Frame).

encode_frame(Record, #state{channel = Channel}) ->
    Encoded = amqp10_framing:encode_bin(Record),
    amqp10_binary_generator:build_frame(Channel, Encoded).

send(Record, #state{socket = Socket} = State) ->
    Frame = encode_frame(Record, State),
    socket_send(Socket, Frame).

send_transfer(Transfer0, Sections0, FooterOpt, MaxMessageSize,
              #state{socket = Socket,
                     channel = Channel,
                     connection_config = Config}) ->
    OutMaxFrameSize = maps:get(outgoing_max_frame_size, Config),
    Transfer = amqp10_framing:encode_bin(Transfer0),
    TransferSize = iolist_size(Transfer),
    Sections = encode_sections(Sections0, FooterOpt),
    SectionsBin = iolist_to_binary(Sections),
    if is_integer(MaxMessageSize) andalso
       MaxMessageSize > 0 andalso
       byte_size(SectionsBin) > MaxMessageSize ->
           {error, message_size_exceeded};
       true ->
           % TODO: this does not take the extended header into account
           % see: 2.3
           MaxPayloadSize = OutMaxFrameSize - TransferSize - ?FRAME_HEADER_SIZE,
           Frames = build_frames(Channel, Transfer0, SectionsBin, MaxPayloadSize, []),
           ok = socket_send(Socket, Frames),
           {ok, length(Frames)}
    end.

encode_sections(Sections, undefined) ->
    [amqp10_framing:encode_bin(S) || S <- Sections];
encode_sections(Sections, FooterOpt) ->
    {Bare, NoBare} = lists:partition(fun is_bare_message_section/1, Sections),
    {FooterL, PreBare} = lists:partition(fun(#'v1_0.footer'{}) ->
                                                 true;
                                            (_) ->
                                                 false
                                         end, NoBare),
    PreBareEncoded = [amqp10_framing:encode_bin(S) || S <- PreBare],
    BareEncoded = [amqp10_framing:encode_bin(S) || S <- Bare],
    {Key, Checksum} = case FooterOpt of
                          crc32 ->
                              {<<"x-opt-crc-32">>, erlang:crc32(BareEncoded)};
                          adler32 ->
                              {<<"x-opt-adler-32">>, erlang:adler32(BareEncoded)}
                      end,
    Ann = {{symbol, Key}, {uint, Checksum}},
    Footer = case FooterL of
                 [] ->
                     #'v1_0.footer'{content = [Ann]};
                 [F = #'v1_0.footer'{content = Content}] ->
                     F#'v1_0.footer'{content = [Ann | Content]}
             end,
    FooterEncoded = amqp10_framing:encode_bin(Footer),
    [PreBareEncoded, BareEncoded, FooterEncoded].

is_bare_message_section(#'v1_0.header'{}) ->
    false;
is_bare_message_section(#'v1_0.delivery_annotations'{}) ->
    false;
is_bare_message_section(#'v1_0.message_annotations'{}) ->
    false;
is_bare_message_section(#'v1_0.footer'{}) ->
    false;
is_bare_message_section(_Section) ->
    true.

send_flow_link(OutHandle,
               #'v1_0.flow'{link_credit = {uint, Credit}} = Flow0, RenewWhenBelow,
               #state{links = Links} = State) ->
    AutoFlow = case RenewWhenBelow of
                   never -> never;
                   _ -> {RenewWhenBelow, Credit}
               end,
    #{OutHandle := #link{output_handle = H,
                         role = receiver,
                         delivery_count = DeliveryCount,
                         available = Available} = Link} = Links,
    Flow1 = Flow0#'v1_0.flow'{
                    handle = uint(H),
                    %% "In the event that the receiving link endpoint has not yet seen the
                    %% initial attach frame from the sender this field MUST NOT be set." [2.7.4]
                    delivery_count = maybe_uint(DeliveryCount),
                    available = uint(Available)},
    Flow = set_flow_session_fields(Flow1, State),
    ok = send(Flow, State),
    State#state{links = Links#{OutHandle =>
                               Link#link{link_credit = Credit,
                                         auto_flow = AutoFlow}}}.

send_flow_session(State) ->
    Flow = set_flow_session_fields(#'v1_0.flow'{}, State),
    ok = send(Flow, State).

set_flow_session_fields(Flow, #state{next_incoming_id = NID,
                                     incoming_window = IW,
                                     next_outgoing_id = NOI}) ->
    Flow#'v1_0.flow'{
           %% "This value MUST be set if the peer has received the begin
           %% frame for the session, and MUST NOT be set if it has not." [2.7.4]
           next_incoming_id = maybe_uint(NID),
           %% IncomingWindow0 can be negative when the sending server overshoots our window.
           %% We must set a floor of 0 in the FLOW frame because field incoming-window is an uint.
           incoming_window = uint(max(0, IW)),
           next_outgoing_id = uint(NOI),
           outgoing_window = ?UINT_OUTGOING_WINDOW}.

build_frames(Channel, Trf, Bin, MaxPayloadSize, Acc)
  when byte_size(Bin) =< MaxPayloadSize ->
    T = amqp10_framing:encode_bin(Trf#'v1_0.transfer'{more = false}),
    Frame = amqp10_binary_generator:build_frame(Channel, [T, Bin]),
    lists:reverse([Frame | Acc]);
build_frames(Channel, Trf, Payload, MaxPayloadSize, Acc) ->
    <<Bin:MaxPayloadSize/binary, Rest/binary>> = Payload,
    T = amqp10_framing:encode_bin(Trf#'v1_0.transfer'{more = true}),
    Frame = amqp10_binary_generator:build_frame(Channel, [T, Bin]),
    build_frames(Channel, Trf, Rest, MaxPayloadSize, [Frame | Acc]).

make_source(#{role := {sender, _}}) ->
    #'v1_0.source'{};
make_source(#{role := {receiver, Source, _Pid},
              filter := Filter}) ->
    Durable = translate_terminus_durability(maps:get(durable, Source, none)),
    Dynamic = maps:get(dynamic, Source, false),
    TranslatedFilter = translate_filters(Filter),
    #'v1_0.source'{address = make_address(Source),
                   durable = {uint, Durable},
                   dynamic = Dynamic,
                   filter = TranslatedFilter,
                   capabilities = make_capabilities(Source)}.

make_target(#{role := {receiver, _Source, _Pid}}) ->
    #'v1_0.target'{};
make_target(#{role := {sender, Target}}) ->
    Durable = translate_terminus_durability(maps:get(durable, Target, none)),
    Dynamic = maps:get(dynamic, Target, false),
    #'v1_0.target'{address = make_address(Target),
                   durable = {uint, Durable},
                   dynamic = Dynamic,
                   capabilities = make_capabilities(Target)}.

make_address(#{address := Addr}) ->
    if is_binary(Addr) ->
           {utf8, Addr};
       is_atom(Addr) ->
           Addr
    end.

make_capabilities(#{capabilities := Caps0}) ->
    Caps = [{symbol, C} || C <- Caps0],
    {array, symbol, Caps};
make_capabilities(_) ->
    undefined.

max_message_size(#{max_message_size := Size})
  when is_integer(Size) andalso
       Size > 0 ->
    {ulong, Size};
max_message_size(_) ->
    undefined.

translate_terminus_durability(none) -> 0;
translate_terminus_durability(configuration) -> 1;
translate_terminus_durability(unsettled_state) -> 2.

translate_filters(Filters)
  when map_size(Filters) =:= 0 ->
    undefined;
translate_filters(Filters) ->
    {map,
     maps:fold(
       fun
           (<<"apache.org:legacy-amqp-headers-binding:map">> = K, V, Acc) when is_map(V) ->
               %% special case conversion
               Key = sym(K),
               [{Key, {described, Key, translate_legacy_amqp_headers_binding(V)}} | Acc];
          (K, V, Acc) when is_binary(K) ->
               %% try treat any filter value generically
               Key = sym(K),
               Value = filter_value_type(V),
               [{Key, {described, Key, Value}} | Acc]
       end, [], Filters)}.

filter_value_type(V) when is_binary(V) ->
    %% this is clearly not always correct
    {utf8, V};
filter_value_type(V)
  when is_integer(V) andalso V >= 0 ->
    {uint, V};
filter_value_type(VList) when is_list(VList) ->
    {list, [filter_value_type(V) || V <- VList]};
filter_value_type({T, _} = V) when is_atom(T) ->
    %% looks like an already tagged type, just pass it through
    V.

% https://people.apache.org/~rgodfrey/amqp-1.0/apache-filters.html
translate_legacy_amqp_headers_binding(LegacyHeaders) ->
    {map,
     maps:fold(
       fun(<<"x-match">> = K, <<"any">> = V, Acc) ->
               [{{utf8, K}, {utf8, V}} | Acc];
          (<<"x-match">> = K, <<"all">> = V, Acc) ->
               [{{utf8, K}, {utf8, V}} | Acc];
          (<<"x-", _/binary>>, _, Acc) ->
               Acc;
          (K, V, Acc) ->
               [{{utf8, K}, filter_value_type(V)} | Acc]
       end, [], LegacyHeaders)}.

send_detach(Send, {detach, OutHandle}, _From, State = #state{links = Links}) ->
    case Links of
        #{OutHandle := #link{input_handle = undefined}} ->
            % Link = stash_link_request(Link0, From, Msg),
            % not fully attached yet - stash the request for later processing
            {{error, half_attached}, State};
        #{OutHandle := Link} ->
            Detach = #'v1_0.detach'{handle = uint(OutHandle),
                                    closed = true},
            ok = Send(Detach, State),
            Links1 = Links#{OutHandle => Link#link{state = detach_sent}},
            {ok, State#state{links = Links1}};
        _ ->
            {{error, link_not_found}, State}
    end.

detach_with_error_cond(Link = #link{output_handle = OutHandle}, State, Cond, Description) ->
    Err = #'v1_0.error'{condition = Cond,
                        description = Description},
    Detach = #'v1_0.detach'{handle = uint(OutHandle),
                            closed = true,
                            error = Err},
    ok = send(Detach, State),
    Link#link{state = detach_sent}.

send_attach(Send, #{name := Name, role := RoleTuple} = Args, {FromPid, _},
            #state{next_link_handle = OutHandle0, links = Links,
             link_index = LinkIndex} = State) ->

    Source = make_source(Args),
    Target = make_target(Args),
    Properties = amqp10_client_types:make_properties(Args),

    {LinkTarget, InitialDeliveryCount, MaxMessageSize} =
    case RoleTuple of
        {receiver, _, Pid} ->
            {{pid, Pid}, undefined, max_message_size(Args)};
        {sender, #{address := TargetAddr}} ->
            {TargetAddr, uint(?INITIAL_DELIVERY_COUNT), undefined}
    end,

    {OutHandle, NextLinkHandle} = case Args of
                                      #{handle := Handle} ->
                                          %% Client app provided link handle.
                                          %% Really only meant for integration tests.
                                          {Handle, OutHandle0};
                                      _ ->
                                          {OutHandle0, OutHandle0 + 1}
                                  end,
    Role = element(1, RoleTuple),
    % create attach performative
    Attach = #'v1_0.attach'{name = {utf8, Name},
                            role = role_to_boolean(Role),
                            handle = {uint, OutHandle},
                            source = Source,
                            properties = Properties,
                            initial_delivery_count = InitialDeliveryCount,
                            snd_settle_mode = snd_settle_mode(Args),
                            rcv_settle_mode = rcv_settle_mode(Args),
                            target = Target,
                            max_message_size = MaxMessageSize},
    ok = Send(Attach, State),

    Ref = make_link_ref(Role, self(), OutHandle),
    Link = #link{name = Name,
                 ref = Ref,
                 output_handle = OutHandle,
                 state = attach_sent,
                 role = Role,
                 notify = FromPid,
                 auto_flow = never,
                 target = LinkTarget,
                 delivery_count = unpack(InitialDeliveryCount),
                 max_message_size = unpack(MaxMessageSize),
                 footer_opt = maps:get(footer_opt, Args, undefined)},

    {State#state{links = Links#{OutHandle => Link},
                 next_link_handle = NextLinkHandle,
                 link_index = LinkIndex#{{Role, Name} => OutHandle}}, Ref}.

-spec handle_session_flow(#'v1_0.flow'{}, #state{}) -> #state{}.
handle_session_flow(#'v1_0.flow'{next_incoming_id = MaybeNII,
                                 next_outgoing_id = {uint, NOI},
                                 incoming_window = {uint, InWin},
                                 outgoing_window = {uint, OutWin}},
       #state{next_outgoing_id = OurNOI} = State) ->
    NII = case MaybeNII of
              {uint, N} -> N;
              undefined -> ?INITIAL_OUTGOING_TRANSFER_ID
          end,
    RemoteIncomingWindow = diff(add(NII, InWin), OurNOI), % see: 2.5.6
    State#state{next_incoming_id = NOI,
                remote_incoming_window = RemoteIncomingWindow,
                remote_outgoing_window = OutWin}.


-spec handle_link_flow(#'v1_0.flow'{}, #link{}) -> {ok | send_flow, #link{}}.
handle_link_flow(#'v1_0.flow'{drain = true,
                              link_credit = {uint, TheirCredit}},
                 Link = #link{role = sender,
                              delivery_count = OurDC,
                              available = 0}) ->
    {send_flow, Link#link{link_credit = 0,
                          delivery_count = add(OurDC, TheirCredit)}};
handle_link_flow(#'v1_0.flow'{delivery_count = MaybeTheirDC,
                              link_credit = {uint, TheirCredit}},
                 Link = #link{role = sender,
                              delivery_count = OurDC}) ->
    TheirDC = case MaybeTheirDC of
                  {uint, DC} -> DC;
                  undefined -> ?INITIAL_DELIVERY_COUNT
              end,
    LinkCredit = amqp10_util:link_credit_snd(TheirDC, TheirCredit, OurDC),
    {ok, Link#link{link_credit = LinkCredit}};
handle_link_flow(#'v1_0.flow'{delivery_count = TheirDC,
                              link_credit = {uint, TheirCredit},
                              available = Available,
                              drain = Drain0},
                 Link0 = #link{role = receiver}) ->
    Drain = default(Drain0, false),
    Link = case Drain andalso TheirCredit =:= 0 of
               true ->
                   notify_credit_exhausted(Link0),
                   Link0#link{delivery_count = unpack(TheirDC),
                              link_credit = 0,
                              available = unpack(Available),
                              drain = Drain};
               false ->
                   Link0#link{delivery_count = unpack(TheirDC),
                              available = unpack(Available),
                              drain = Drain}
           end,
    {ok, Link}.

-spec find_link_by_input_handle(input_handle(), #state{}) ->
    {ok, #link{}} | not_found.
find_link_by_input_handle(InHandle, #state{link_handle_index = LHI,
                                           links = Links}) ->
    case LHI of
        #{InHandle := OutHandle} ->
            case Links of
                #{OutHandle := Link} ->
                    {ok, Link};
                _ -> not_found
            end;
        _ -> not_found
    end.

with_link(InHandle, State, Fun) ->
    case find_link_by_input_handle(InHandle, State) of
        {ok, Link} ->
            Fun(Link, State);
        not_found ->
            % end session with errant-link
            ok = send_end(State, ?V_1_0_SESSION_ERROR_ERRANT_LINK),
            {next_state, end_sent, State}
    end.

maybe_uint(undefined) -> undefined;
maybe_uint(Int) -> uint(Int).

uint(Int) -> {uint, Int}.

unpack(X) -> amqp10_client_types:unpack(X).

snd_settle_mode(#{snd_settle_mode := unsettled}) -> {ubyte, 0};
snd_settle_mode(#{snd_settle_mode := settled}) -> {ubyte, 1};
snd_settle_mode(#{snd_settle_mode := mixed}) -> {ubyte, 2};
snd_settle_mode(_) -> undefined.

rcv_settle_mode(#{rcv_settle_mode := first}) -> {ubyte, 0};
rcv_settle_mode(#{rcv_settle_mode := second}) -> {ubyte, 1};
rcv_settle_mode(_) -> undefined.

% certain amqp10 brokers (IBM MQ) return an undefined delivery state
% when the link is detached before settlement is sent
% TODO: work out if we can assume accepted
translate_delivery_state(undefined) -> undefined;
translate_delivery_state(#'v1_0.accepted'{}) -> accepted;
translate_delivery_state(#'v1_0.rejected'{error = undefined}) -> rejected;
translate_delivery_state(#'v1_0.rejected'{error = Error}) -> {rejected, Error};
translate_delivery_state(#'v1_0.modified'{}) -> modified;
translate_delivery_state(#'v1_0.released'{}) -> released;
translate_delivery_state(#'v1_0.received'{}) -> received;
translate_delivery_state(accepted) -> #'v1_0.accepted'{};
translate_delivery_state(rejected) -> #'v1_0.rejected'{};
translate_delivery_state(modified) -> #'v1_0.modified'{};
translate_delivery_state({modified,
                          DeliveryFailed,
                          UndeliverableHere,
                          MessageAnnotations}) ->
    MA = translate_message_annotations(MessageAnnotations),
    #'v1_0.modified'{delivery_failed = DeliveryFailed,
                     undeliverable_here = UndeliverableHere,
                     message_annotations = MA};
translate_delivery_state(released) -> #'v1_0.released'{};
translate_delivery_state(received) -> #'v1_0.received'{}.

translate_role(receiver) -> true.

maybe_notify_link_credit(#link{role = sender,
                               link_credit = 0},
                         #link{role = sender,
                               link_credit = NewCredit} = NewLink)
  when NewCredit > 0 ->
    notify_link(NewLink, credited);
maybe_notify_link_credit(_Old, _New) ->
    ok.

notify_link_attached(Link, Perf, #state{connection_config = Cfg}) ->
    What = case Cfg of
               #{notify_with_performative := true} ->
                   {attached, Perf};
               _ ->
                   attached
           end,
    notify_link(Link, What).

notify_link_detached(Link,
                     Perf = #'v1_0.detach'{error = Err},
                     #state{connection_config = Cfg}) ->
    Reason = case Cfg of
                 #{notify_with_performative := true} ->
                     Perf;
                 _ ->
                     reason(Err)
             end,
    notify_link(Link, {detached, Reason}).

notify_link(#link{notify = Pid, ref = Ref}, What) ->
    Evt = {amqp10_event, {link, Ref, What}},
    Pid ! Evt,
    ok.

notify_session_begun(Perf, #state{notify = Pid,
                                  connection_config = Cfg}) ->
    Evt = case Cfg of
              #{notify_with_performative := true} ->
                  {begun, Perf};
              _ ->
                  begun
          end,
    Pid ! amqp10_session_event(Evt),
    ok.

notify_session_ended(Perf = #'v1_0.end'{error = Err},
                     #state{notify = Pid,
                            connection_config = Cfg}) ->
    Reason = case Cfg of
                 #{notify_with_performative := true} ->
                     Perf;
                 _ ->
                     reason(Err)
             end,
    Pid ! amqp10_session_event({ended, Reason}),
    ok.

notify_disposition(Pid, DeliveryStateDeliveryTag) ->
    Pid ! {amqp10_disposition, DeliveryStateDeliveryTag},
    ok.

book_transfer_send(Num, #link{output_handle = Handle} = Link,
                   #state{next_outgoing_id = NextOutgoingId,
                          outgoing_delivery_id = DeliveryId,
                          remote_incoming_window = RIW,
                          links = Links} = State) ->
    State#state{next_outgoing_id = add(NextOutgoingId, Num),
                outgoing_delivery_id = add(DeliveryId, 1),
                remote_incoming_window = RIW-Num,
                links = Links#{Handle => book_link_transfer_send(Link)}}.

book_partial_transfer_received(#state{next_incoming_id = NID,
                                      incoming_window = IW,
                                      remote_outgoing_window = ROW} = State0) ->
    State = State0#state{next_incoming_id = add(NID, 1),
                         incoming_window = IW - 1,
                         remote_outgoing_window = ROW - 1},
    maybe_widen_incoming_window(State).

book_transfer_received(State = #state{connection_config =
                                      #{transfer_limit_margin := Margin}},
                       #link{link_credit = Margin} = Link) ->
    {transfer_limit_exceeded, Link, State};
book_transfer_received(#state{next_incoming_id = NID,
                              incoming_window = IW,
                              remote_outgoing_window = ROW,
                              links = Links} = State0,
                       #link{output_handle = OutHandle,
                             delivery_count = DC,
                             link_credit = LC,
                             available = Avail} = Link) ->
    Link1 = Link#link{delivery_count = add(DC, 1),
                      link_credit = LC - 1,
                      %% "the receiver MUST maintain a floor of zero in its
                      %% calculation of the value of available" [2.6.7]
                      available = max(0, Avail - 1)},
    State1 = State0#state{links = Links#{OutHandle => Link1},
                          next_incoming_id = add(NID, 1),
                          incoming_window = IW - 1,
                          remote_outgoing_window = ROW - 1},
    State = maybe_widen_incoming_window(State1),
    case Link1 of
        #link{link_credit = 0,
              auto_flow = never} ->
            {credit_exhausted, Link1, State};
        _ ->
            {ok, Link1, State}
    end.

maybe_widen_incoming_window(
  State0 = #state{incoming_window = IncomingWindow,
                  auto_flow = {RenewWhenBelow, NewWindowSize}})
  when IncomingWindow < RenewWhenBelow ->
    State = State0#state{incoming_window = NewWindowSize},
    send_flow_session(State),
    State;
maybe_widen_incoming_window(State) ->
    State.

auto_flow(#link{link_credit = LC,
                auto_flow = {RenewWhenBelow, Credit},
                output_handle = OutHandle,
                incoming_unsettled = Unsettled},
          State)
  when LC + map_size(Unsettled) < RenewWhenBelow ->
    send_flow_link(OutHandle,
                   #'v1_0.flow'{link_credit = {uint, Credit}},
                   RenewWhenBelow, State);
auto_flow(_, State) ->
    State.

update_link(Link = #link{output_handle = OutHandle},
            State = #state{links = Links}) ->
            State#state{links = Links#{OutHandle => Link}}.

book_link_transfer_send(Link = #link{link_credit = LC,
                                     delivery_count = DC}) ->
    Link#link{link_credit = LC - 1,
              delivery_count = add(DC, 1)}.

append_partial_transfer(Transfer, Payload,
                        #link{partial_transfers = undefined} = Link) ->
    Link#link{partial_transfers = {Transfer, [Payload]}};
append_partial_transfer(_Transfer, Payload,
                        #link{partial_transfers = {T, Payloads}} = Link) ->
    Link#link{partial_transfers = {T, [Payload | Payloads]}}.

complete_partial_transfer(Transfer, Payload,
                          #link{partial_transfers = undefined} = Link) ->
    {Transfer, Payload, Link};
complete_partial_transfer(_Transfer, Payload,
                          #link{partial_transfers = {T, Payloads}} = Link) ->
    {T, iolist_to_binary(lists:reverse([Payload | Payloads])),
     Link#link{partial_transfers = undefined}}.

decode_as_msg(Transfer, Payload, undefined) ->
    Sections = amqp10_framing:decode_bin(Payload),
    {ok, amqp10_msg:from_amqp_records([Transfer | Sections])};
decode_as_msg(Transfer, Payload, FooterOpt) ->
    PosSections = decode_sections([], Payload, byte_size(Payload), 0),
    Sections = lists:map(fun({_Pos, S}) -> S end, PosSections),
    Msg = amqp10_msg:from_amqp_records([Transfer | Sections]),
    OkMsg = {ok, Msg},
    case lists:last(PosSections) of
        {PosFooter, #'v1_0.footer'{content = Content}}
          when is_list(Content) ->
            Key = case FooterOpt of
                      crc32 -> <<"x-opt-crc-32">>;
                      adler32 -> <<"x-opt-adler-32">>
                  end,
            case lists:search(fun({{symbol, K}, {uint, _Checksum}})
                                    when K =:= Key ->
                                      true;
                                 (_) ->
                                      false
                              end, Content) of
                {value, {{symbol, Key}, {uint, Expected}}} ->
                    {value, {PosBare, _}} = lists:search(fun({_Pos, S}) ->
                                                                 is_bare_message_section(S)
                                                         end, PosSections),
                    BareBin = binary_part(Payload, PosBare, PosFooter - PosBare),
                    case erlang:FooterOpt(BareBin) of
                        Expected -> OkMsg;
                        Actual -> {checksum_error, Expected, Actual}
                    end;
                false ->
                    OkMsg
            end;
        _ ->
            OkMsg
    end.

decode_sections(PosSections, _Payload, PayloadSize, PayloadSize) ->
    lists:reverse(PosSections);
decode_sections(PosSections, Payload, PayloadSize, Parsed)
  when PayloadSize > Parsed ->
    Bin = binary_part(Payload, Parsed, PayloadSize - Parsed),
    {Described, NumBytes} = amqp10_binary_parser:parse(Bin),
    Section = amqp10_framing:decode(Described),
    decode_sections([{Parsed, Section} | PosSections],
                    Payload,
                    PayloadSize,
                    Parsed + NumBytes).

amqp10_session_event(Evt) ->
    {amqp10_event, {session, self(), Evt}}.

socket_send(Sock, Data) ->
    case amqp10_client_socket:send(Sock, Data) of
        ok ->
            ok;
        {error, _Reason} ->
            throw({stop, normal})
    end.

%% Only notify of credit exhaustion when not using auto flow.
notify_credit_exhausted(Link = #link{auto_flow = never}) ->
    ok = notify_link(Link, credit_exhausted);
notify_credit_exhausted(_Link) ->
    ok.

-spec make_link_ref(link_role(), pid(), output_handle()) ->
    link_ref().
make_link_ref(Role, Session, Handle) ->
    #link_ref{role = Role, session = Session, link_handle = Handle}.

translate_message_annotations(MA)
  when map_size(MA) > 0 ->
    {map, maps:fold(fun(K, V, Acc) ->
                            [{sym(K), amqp10_client_types:infer(V)} | Acc]
                    end, [], MA)};
translate_message_annotations(_MA) ->
    undefined.

sym(B) when is_binary(B) -> {symbol, B};
sym(B) when is_list(B) -> {symbol, list_to_binary(B)};
sym(B) when is_atom(B) -> {symbol, atom_to_binary(B, utf8)}.

reason(undefined) -> normal;
reason(Other) -> Other.

role_to_boolean(sender) ->
    ?AMQP_ROLE_SENDER;
role_to_boolean(receiver) ->
    ?AMQP_ROLE_RECEIVER.

boolean_to_role(?AMQP_ROLE_SENDER) ->
    sender;
boolean_to_role(?AMQP_ROLE_RECEIVER) ->
    receiver.

default(undefined, Default) -> Default;
default(Thing, _Default) -> Thing.

format_status(Status = #{data := Data0}) ->
    #state{channel = Channel,
           remote_channel = RemoteChannel,
           next_incoming_id = NextIncomingId,
           incoming_window = IncomingWindow,
           auto_flow = SessionAutoFlow,
           next_outgoing_id = NextOutgoingId,
           remote_incoming_window = RemoteIncomingWindow,
           remote_outgoing_window = RemoteOutgoingWindow,
           reader = Reader,
           socket = Socket,
           links = Links0,
           link_index = LinkIndex,
           link_handle_index = LinkHandleIndex,
           next_link_handle = NextLinkHandle,
           early_attach_requests = EarlyAttachRequests,
           connection_config = ConnectionConfig,
           outgoing_delivery_id = OutgoingDeliveryId,
           outgoing_unsettled = OutgoingUnsettled,
           notify = Notify
          } = Data0,
    Links = maps:map(
              fun(_OutputHandle,
                  #link{name = Name,
                        ref = Ref,
                        state = State,
                        notify = LinkNotify,
                        output_handle = OutputHandle,
                        input_handle = InputHandle,
                        role = Role,
                        target = Target,
                        max_message_size = MaxMessageSize,
                        delivery_count = DeliveryCount,
                        link_credit = LinkCredit,
                        available = Available,
                        drain = Drain,
                        partial_transfers = PartialTransfers0,
                        auto_flow = AutoFlow,
                        incoming_unsettled = IncomingUnsettled,
                        footer_opt = FooterOpt
                       }) ->
                      PartialTransfers = case PartialTransfers0 of
                                             undefined ->
                                                 0;
                                             {#'v1_0.transfer'{}, Binaries} ->
                                                 length(Binaries)
                                         end,
                      #{name => Name,
                        ref => Ref,
                        state => State,
                        notify => LinkNotify,
                        output_handle => OutputHandle,
                        input_handle => InputHandle,
                        role => Role,
                        target => Target,
                        max_message_size => MaxMessageSize,
                        delivery_count => DeliveryCount,
                        link_credit => LinkCredit,
                        available => Available,
                        drain => Drain,
                        partial_transfers => PartialTransfers,
                        auto_flow => AutoFlow,
                        incoming_unsettled => maps:size(IncomingUnsettled),
                        footer_opt => FooterOpt
                       }
              end, Links0),
    Data = #{channel => Channel,
             remote_channel => RemoteChannel,
             next_incoming_id => NextIncomingId,
             incoming_window => IncomingWindow,
             auto_flow => SessionAutoFlow,
             next_outgoing_id => NextOutgoingId,
             remote_incoming_window => RemoteIncomingWindow,
             remote_outgoing_window => RemoteOutgoingWindow,
             reader => Reader,
             socket => Socket,
             links => Links,
             link_index => LinkIndex,
             link_handle_index => LinkHandleIndex,
             next_link_handle => NextLinkHandle,
             early_attach_requests => length(EarlyAttachRequests),
             connection_config => maps:remove(sasl, ConnectionConfig),
             outgoing_delivery_id => OutgoingDeliveryId,
             outgoing_unsettled => maps:size(OutgoingUnsettled),
             notify => Notify},
    Status#{data := Data}.

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-include_lib("kernel/include/logger.hrl").

handle_session_flow_test() ->
    % see spec section: 2.5.6 for logic
    Flow = #'v1_0.flow'{next_incoming_id = {uint, 50},
                        next_outgoing_id = {uint, 42},
                        incoming_window = {uint, 1000},
                        outgoing_window = {uint, 2000}},
    State0 = #state{next_outgoing_id = 51},
    State = handle_session_flow(Flow, State0),
    42 = State#state.next_incoming_id,
    2000 = State#state.remote_outgoing_window,
    50 + 1000 - 51 = State#state.remote_incoming_window.

handle_session_flow_pre_begin_test() ->
    % see spec section: 2.5.6 for logic
    Flow = #'v1_0.flow'{next_incoming_id = undefined,
                        next_outgoing_id = {uint, 42},
                        incoming_window = {uint, 1000},
                        outgoing_window = {uint, 2000}},
    State0 = #state{next_outgoing_id = 51},
    State = handle_session_flow(Flow, State0),
    42 = State#state.next_incoming_id,
    2000 = State#state.remote_outgoing_window,
    % using serial number arithmetic:
    % ?INITIAL_OUTGOING_TRANSFER_ID + 1000 = 998
    ?assertEqual(998 - 51, State#state.remote_incoming_window).

handle_link_flow_sender_test() ->
    DeliveryCountRcv = 55,
    DeliveryCountSnd = DeliveryCountRcv + 2,
    LinkCreditRcv = 42,
    Link = #link{role = sender,
                 output_handle = 99,
                 link_credit = 0,
                 delivery_count = DeliveryCountSnd},
    Flow = #'v1_0.flow'{handle = {uint, 45},
                        link_credit = {uint, LinkCreditRcv},
                        delivery_count = {uint, DeliveryCountRcv}
                       },
    {ok, Outcome} = handle_link_flow(Flow, Link),
    % see section 2.6.7
    ?assertEqual(DeliveryCountRcv + LinkCreditRcv - DeliveryCountSnd,
                 Outcome#link.link_credit),

    % receiver does not yet know the delivery_count
    {ok, Outcome2} = handle_link_flow(Flow#'v1_0.flow'{delivery_count = undefined},
                                      Link),
    % using serial number arithmetic:
    % ?INITIAL_DELIVERY_COUNT + LinkCreditRcv - DeliveryCountSnd = -18
    % but we maintain a floor of zero
    ?assertEqual(0, Outcome2#link.link_credit).

handle_link_flow_sender_drain_test() ->
    Handle = 45,
    SndDeliveryCount = 55,
    RcvLinkCredit = 42,
    Link = #link{role = sender, output_handle = 99,
                 available = 0, link_credit = 20,
                 delivery_count = SndDeliveryCount},
    Flow = #'v1_0.flow'{handle = {uint, Handle},
                        link_credit = {uint, RcvLinkCredit},
                        drain = true},
    {send_flow, Outcome} = handle_link_flow(Flow, Link),
    0 = Outcome#link.link_credit,
    ExpectedDC = SndDeliveryCount + RcvLinkCredit,
    ExpectedDC = Outcome#link.delivery_count.


handle_link_flow_receiver_test() ->
    Handle = 45,
    DeliveryCount = 55,
    SenderDC = 57,
    Link = #link{role = receiver, output_handle = 99,
                 link_credit = 0, delivery_count = DeliveryCount},
    Flow = #'v1_0.flow'{handle = {uint, Handle},
                        delivery_count = {uint, SenderDC},
                        available = 99,
                        drain = true, % what to do?
                        link_credit = {uint, 0}
                       },
    {ok, Outcome} = handle_link_flow(Flow, Link),
    % see section 2.6.7
    99 = Outcome#link.available,
    true = Outcome#link.drain,
    SenderDC = Outcome#link.delivery_count. % maintain delivery count

translate_filters_empty_map_test() ->
    undefined = translate_filters(#{}).

translate_filters_legacy_amqp_direct_binding_filter_test() ->
    {map,
        [{
            {symbol,<<"apache.org:legacy-amqp-direct-binding:string">>},
            {described, {symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, {utf8,<<"my topic">>}}
        }]
    } = translate_filters(#{<<"apache.org:legacy-amqp-direct-binding:string">> => <<"my topic">>}).

translate_filters_legacy_amqp_topic_binding_filter_test() ->
    {map,
        [{
            {symbol, <<"apache.org:legacy-amqp-topic-binding:string">>},
            {described, {symbol, <<"apache.org:legacy-amqp-topic-binding:string">>}, {utf8, <<"*.stock.#">>}}
        }]
    } = translate_filters(#{<<"apache.org:legacy-amqp-topic-binding:string">> => <<"*.stock.#">>}).

translate_filters_legacy_amqp_headers_binding_filter_test() ->
    {map,
        [{
            {symbol, <<"apache.org:legacy-amqp-headers-binding:map">>},
            {described, {symbol, <<"apache.org:legacy-amqp-headers-binding:map">>},
             {map, [
                    {{utf8, <<"x-match">>}, {utf8, <<"all">>}},
                    {{utf8, <<"foo">>}, {utf8, <<"bar">>}},
                    {{utf8, <<"bar">>}, {utf8, <<"baz">>}}
                ]
            }}
        }]
    } = translate_filters(#{<<"apache.org:legacy-amqp-headers-binding:map">> => #{
            <<"x-match">> => <<"all">>,
            <<"x-something">> => <<"ignored">>,
            <<"bar">> => <<"baz">>,
            <<"foo">> => <<"bar">>
        }}).

translate_filters_legacy_amqp_no_local_filter_test() ->
    {map,
        [{
            {symbol, <<"apache.org:no-local-filter:list">>},
            {described, {symbol, <<"apache.org:no-local-filter:list">>},
             {list, [{utf8, <<"foo">>}, {utf8, <<"bar">>}]}}
        }]
    } = translate_filters(#{<<"apache.org:no-local-filter:list">> => [<<"foo">>, <<"bar">>]}).

translate_filters_selector_filter_test() ->
    {map,
        [{
            {symbol, <<"apache.org:selector-filter:string">>},
            {described, {symbol, <<"apache.org:selector-filter:string">>},
             {utf8, <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}}
        }]
    } = translate_filters(#{<<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}).

translate_filters_multiple_filters_test() ->
    {map,
        [
            {{symbol, <<"apache.org:selector-filter:string">>},
             {described, {symbol, <<"apache.org:selector-filter:string">>},
              {utf8, <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}}},
            {{symbol, <<"apache.org:legacy-amqp-direct-binding:string">>},
             {described, {symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, {utf8,<<"my topic">>}}}
        ]
    } = translate_filters(#{
            <<"apache.org:legacy-amqp-direct-binding:string">> => <<"my topic">>,
            <<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>
        }).
-endif.