Rename unconfirmed to incoming_unsettled_map

to better match the AMQP spec terminology.

Author: ansd

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -9,7 +9,7 @@
 
 -behaviour(gen_server).
 
--include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
 -include("rabbit_amqp1_0.hrl").
 
 -define(MAX_SESSION_WINDOW_SIZE, 65_535).
@@ -26,7 +26,7 @@
                    ?V_1_0_SYMBOL_MODIFIED]).
 
 %% Just make these constant for the time being.
--define(INCOMING_CREDIT, 65536).
+-define(INCOMING_CREDIT, 65_536).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 
@@ -40,7 +40,6 @@
          get_info/1]).
 -export([init/1,
          terminate/2,
-         code_change/3,
          handle_call/3,
          handle_cast/2, 
          handle_info/2]).
@@ -53,17 +52,16 @@
           exchange :: rabbit_exchange:name(),
           routing_key :: undefined | rabbit_types:routing_key(),
           delivery_id :: undefined | delivery_number(),
-          delivery_count = 0,
+          delivery_count = 0 :: sequence_no(),
           send_settle_mode = undefined,
           recv_settle_mode = undefined,
           credit_used = ?INCOMING_CREDIT div 2,
           msg_acc = []}).
 
 -record(outgoing_link, {
           queue :: undefined | rabbit_misc:resource_name(),
-          delivery_count = 0,
-          %% TODO below field is not needed?
-          send_settled}).
+          delivery_count = 0 :: sequence_no(),
+          send_settled :: boolean()}).
 
 -record(outgoing_unsettled, {
           %% The queue sent us this consumer scoped sequence number.
@@ -85,8 +83,8 @@
 
 %%TODO put rarely used fields into separate #cfg{}
 -record(state, {frame_max,
-                reader_pid,
-                writer_pid,
+                reader_pid :: pid(),
+                writer_pid :: pid(),
                 %% These messages were received from queues thanks to sufficient link credit.
                 %% However, they are buffered here due to session flow control before being sent to the client.
                 pending_transfers = queue:new() :: queue:queue(#pending_transfer{}),
@@ -103,15 +101,14 @@
                 outgoing_window_max,
                 next_publish_id, %% the 0-9-1-side counter for confirms
                 next_delivery_id = 0 :: delivery_number(),
-                incoming_unsettled_map, %%TODO delete
-                outgoing_unsettled_map = gb_trees:empty() :: gb_trees:tree(delivery_number(), #outgoing_unsettled{}),
-                queue_states = rabbit_queue_type:init() :: rabbit_queue_type:state(),
                 %% TRANSFER delivery IDs published to queues but not yet confirmed by queues
                 %% TODO Use a different data structure because
                 %% 1. we don't need to record exchanges since we don't emit channel stats,
                 %% 2. mixed mode can result in large gaps across delivery_ids that need to be confirmed. Use a tree?
                 %% 3. handle wrap around of 32-bit RFC-1982 serial number
-                unconfirmed = rabbit_confirms:init() :: rabbit_confirms:state()
+                incoming_unsettled_map = rabbit_confirms:init() :: rabbit_confirms:state(),
+                outgoing_unsettled_map = gb_trees:empty() :: gb_trees:tree(delivery_number(), #outgoing_unsettled{}),
+                queue_states = rabbit_queue_type:init() :: rabbit_queue_type:state()
                 %% TRANSFER delivery IDs confirmed by queues but yet to be sent to the client
                 %%TODO accumulate confirms and send DISPOSITIONs after processing the mailbox
                 %%(see rabbit_channel:noreply_coalesce/1
@@ -138,63 +135,18 @@ init({Channel, ReaderPid, WriterPid, User, Vhost, FrameMax}) ->
                 user = User,
                 vhost = Vhost,
                 channel_num            = Channel,
-                next_publish_id        = 0,
-                incoming_unsettled_map = gb_trees:empty()
+                next_publish_id        = 0
                }}.
 
 terminate(_Reason, _State) ->
     ok.
 
-code_change(_OldVsn, State, _Extra) ->
-    {ok, State}.
-
 handle_call(info, _From, #state{reader_pid = ReaderPid} = State) ->
     Info = [{reader, ReaderPid}],
     {reply, Info, State};
 handle_call(Msg, _From, State) ->
     {reply, {error, not_understood, Msg}, State}.
 
-handle_info(#'basic.consume_ok'{}, State) ->
-    %% Handled above
-    {noreply, State};
-
-handle_info(#'basic.cancel_ok'{}, State) ->
-    %% just ignore this for now,
-    %% At some point we should send the detach here but then we'd need to track
-    %% consumer tags -> link handle somewhere
-    {noreply, State};
-
-%% A message from the queue saying that there are no more messages
-% handle_info(#'basic.credit_drained'{consumer_tag = CTag} = CreditDrained,
-%             State = #state{writer_pid = WriterPid,
-%                            session = Session}) ->
-%     Handle = ctag_to_handle(CTag),
-%     Link = get({out, Handle}),
-%     {Flow0, Link1} = rabbit_amqp1_0_session:outgoing_link_credit_drained(
-%                       CreditDrained, Handle, Link),
-%     Flow = rabbit_amqp1_0_session:flow_fields(Flow0, Session),
-%     rabbit_amqp1_0_writer:send_command(WriterPid, Flow),
-%     put({out, Handle}, Link1),
-%     {noreply, State};
-
-% handle_info({#'basic.return'{}, {DTag, _Msg}}, State = #state{writer_pid = WriterPid,
-%                                                               session    = Session}) ->
-%     {Reply, Session1} = rabbit_amqp1_0_session:return(DTag, Session),
-%     case Reply of
-%         undefined ->
-%             ok;
-%         _ ->
-%             rabbit_amqp1_0_writer:send_command(
-%               WriterPid,
-%               rabbit_amqp1_0_session:flow_fields(Reply, Session)
-%              )
-%     end,
-%     {noreply, state(Session1, State)};
-
-% handle_info({#'basic.return'{}, _Msg}, State = #state{session = Session}) ->
-%     rabbit_log:warning("AMQP 1.0 message return without publishing sequence"),
-%     {noreply, state(Session, State)};
-
 handle_info({bump_credit, Msg}, State) ->
     credit_flow:handle_bump_msg(Msg),
     {noreply, State};
@@ -830,12 +782,6 @@ flow(#'v1_0.flow'{next_incoming_id = FlowNextIn0,
             end
     end.
 
-acknowledgement(DeliveryIds, Disposition) ->
-    Disposition#'v1_0.disposition'{first = {uint, hd(DeliveryIds)},
-                                   last = {uint, lists:last(DeliveryIds)},
-                                   settled = true,
-                                   state = #'v1_0.accepted'{}}.
-
 set_delivery_id({uint, D}, #incoming_link{delivery_id = undefined} = Link) ->
     %% "The delivery-id MUST be supplied on the first transfer of a multi-transfer delivery.
     Link#incoming_link{delivery_id = D};
@@ -893,20 +839,43 @@ handle_queue_event({queue_event, QRef, Evt},
             %     rabbit_misc:protocol_error(Type, Reason, ReasonArgs)
     end.
 
+% handle_info({#'basic.return'{}, {DTag, _Msg}}, State = #state{writer_pid = WriterPid,
+%                                                               session    = Session}) ->
+%     {Reply, Session1} = rabbit_amqp1_0_session:return(DTag, Session),
+%     case Reply of
+%         undefined ->
+%             ok;
+%         _ ->
+%             rabbit_amqp1_0_writer:send_command(
+%               WriterPid,
+%               rabbit_amqp1_0_session:flow_fields(Reply, Session)
+%              )
+%     end,
+%     {noreply, state(Session1, State)};
+
+% handle_info({#'basic.return'{}, _Msg}, State = #state{session = Session}) ->
+%     rabbit_log:warning("AMQP 1.0 message return without publishing sequence"),
+%     {noreply, state(Session, State)};
+
 handle_queue_actions(Actions, State0) ->
     {ReplyRev, State} =
     lists:foldl(
-      fun ({settled, QName, DelIds}, {Reply, S0 = #state{unconfirmed = U0}}) ->
+      fun ({settled, QName, DelIds}, {Reply, S0 = #state{incoming_unsettled_map = U0}}) ->
               {ConfirmMXs, U} = rabbit_confirms:confirm(DelIds, QName, U0),
-              S = S0#state{unconfirmed = U},
+              S = S0#state{incoming_unsettled_map = U},
               R = if ConfirmMXs =:= [] ->
                          Reply;
                      ConfirmMXs =/= [] ->
                          ConfirmDelIds = lists:map(fun({Id, _Exchange}) -> Id end, ConfirmMXs),
                          Ids = lists:usort(ConfirmDelIds),
                          %%TODO defer sending confirms as done in rabbit_channel
                          % record_confirms(ConfirmDelIds, S)
-                         Disposition = acknowledgement(Ids, #'v1_0.disposition'{role = ?RECV_ROLE}),
+                         Disposition = #'v1_0.disposition'{
+                                          role = ?RECV_ROLE,
+                                          settled = true,
+                                          state = #'v1_0.accepted'{},
+                                          first = {uint, hd(Ids)},
+                                          last = {uint, lists:last(Ids)}},
                          [Disposition | Reply]
                   end,
               {R, S};
@@ -960,7 +929,7 @@ handle_deliver(ConsumerTag, AckRequired,
             Dtag = if is_integer(MsgId) ->
                           %% delivery-tag must be unique only per link (not per session)
                           <<MsgId:64>>;
-                      MsgId =:= undefined ->
+                      MsgId =:= undefined andalso SendSettled ->
                           %% Both ends of the link will always consider this message settled because
                           %% "the sender will send all deliveries settled to the receiver" [3.8.2].
                           %% Hence, the delivery tag does not have to be unique on this link.
@@ -1039,7 +1008,7 @@ incoming_link_transfer(#'v1_0.transfer'{delivery_id     = DeliveryId0,
                                       send_settle_mode = SSM,
                                       recv_settle_mode = RSM} = Link,
                        #state{queue_states = QStates0,
-                              unconfirmed = U0,
+                              incoming_unsettled_map = U0,
                               next_publish_id = NextPublishId0
                              } = State0) ->
     MsgBin = iolist_to_binary(lists:reverse([MsgPart | MsgAcc])),
@@ -1094,7 +1063,7 @@ incoming_link_transfer(#'v1_0.transfer'{delivery_id     = DeliveryId0,
             {U, Reply0} = process_routing_confirm(Qs, EffectiveSendSettleMode, DeliveryId, XName, U0),
             State1 = State0#state{queue_states = QStates,
                                   next_publish_id = NextPublishId,
-                                  unconfirmed = U},
+                                  incoming_unsettled_map = U},
             {Reply1, State} = handle_queue_actions(Actions, State1),
             {SendFlow, CreditUsed1} = case CreditUsed - 1 of
                                           C when C =< 0 ->