simplify variable_queue:ack's return format

3 files changed, 36 insertions, 43 deletions

diff --git a/include/rabbit_backing_queue_spec.hrl b/include/rabbit_backing_queue_spec.hrl
index f67c6f46d1..6fa34ccc89 100644
--- a/include/rabbit_backing_queue_spec.hrl
+++ b/include/rabbit_backing_queue_spec.hrl
@@ -58,7 +58,7 @@
         (fun ((rabbit_types:message_properties()) -> boolean()), state())
         -> state()).
 -spec(fetch/2 :: (ack_required(), state()) -> {fetch_result(), state()}).
--spec(ack/2 :: ([ack()], state()) -> {[rabbit_guid:guid()], state()}).
+-spec(ack/2 :: ([ack()], state()) -> state()).
 -spec(tx_publish/4 :: (rabbit_types:txn(), rabbit_types:basic_message(),
                        rabbit_types:message_properties(), state()) -> state()).
 -spec(tx_ack/3 :: (rabbit_types:txn(), [ack()], state()) -> state()).
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 53b02a7dbd..960bfd9763 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -818,7 +818,7 @@ handle_call({info, Items}, _From, State) ->
 handle_call(consumers, _From, State) ->
     reply(consumers(State), State);
 
-handle_call({deliver_immediately, Delivery = #delivery{message = Message}},
+handle_call({deliver_immediately, Delivery},
             _From, State) ->
     %% Synchronous, "immediate" delivery mode
     %%
@@ -1011,7 +1011,7 @@ handle_cast({ack, Txn, AckTags, ChPid},
                 case Txn of
                     none -> ChAckTags1 = subtract_acks(ChAckTags, AckTags),
                             NewC = C#cr{acktags = ChAckTags1},
-                            {_AckdGuids, BQS1} = BQ:ack(AckTags, BQS),
+                            BQS1 = BQ:ack(AckTags, BQS),
                             {NewC, State#q{backing_queue_state = BQS1}};
                     _    -> BQS1 = BQ:tx_ack(Txn, AckTags, BQS),
                             {C#cr{txn = Txn},
@@ -1032,7 +1032,7 @@ handle_cast({reject, AckTags, Requeue, ChPid},
             maybe_store_ch_record(C#cr{acktags = ChAckTags1}),
             noreply(case Requeue of
                         true  -> requeue_and_run(AckTags, State);
-                        false -> {_AckdGuids, BQS1} = BQ:ack(AckTags, BQS),
+                        false -> BQS1 = BQ:ack(AckTags, BQS),
                                  State#q{backing_queue_state = BQS1}
                     end)
     end;
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 6264255bcd..aa6d3a8730 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -654,11 +654,9 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
                          persistent_count = PCount1 })}.
 
 ack(AckTags, State) ->
-    {Guids, State1} =
-        ack(fun msg_store_remove/3,
-            fun (_, State0) -> State0 end,
-            AckTags, State),
-    {Guids, a(State1)}.
+    a(ack(fun msg_store_remove/3,
+          fun (_, State0) -> State0 end,
+          AckTags, State)).
 
 tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent }, MsgProps,
            State = #vqstate { durable           = IsDurable,
@@ -708,23 +706,22 @@ tx_commit(Txn, Fun, MsgPropsFun,
        end)}.
 
 requeue(AckTags, MsgPropsFun, State) ->
-    {_Guids, State1} =
-        ack(fun msg_store_release/3,
-            fun (#msg_status { msg = Msg, msg_props = MsgProps }, State1) ->
-                    {_SeqId, State2} = publish(Msg, MsgPropsFun(MsgProps),
-                                               true, false, State1),
-                    State2;
-                ({IsPersistent, Guid, MsgProps}, State1) ->
-                    #vqstate { msg_store_clients = MSCState } = State1,
-                    {{ok, Msg = #basic_message{}}, MSCState1} =
-                        msg_store_read(MSCState, IsPersistent, Guid),
-                    State2 = State1 #vqstate { msg_store_clients = MSCState1 },
-                    {_SeqId, State3} = publish(Msg, MsgPropsFun(MsgProps),
-                                               true, true, State2),
-                    State3
-            end,
-            AckTags, State),
-    a(reduce_memory_use(State1)).
+   a(reduce_memory_use(
+       ack(fun msg_store_release/3,
+           fun (#msg_status { msg = Msg, msg_props = MsgProps }, State1) ->
+                   {_SeqId, State2} = publish(Msg, MsgPropsFun(MsgProps),
+                                              true, false, State1),
+                   State2;
+               ({IsPersistent, Guid, MsgProps}, State1) ->
+                   #vqstate { msg_store_clients = MSCState } = State1,
+                   {{ok, Msg = #basic_message{}}, MSCState1} =
+                       msg_store_read(MSCState, IsPersistent, Guid),
+                   State2 = State1 #vqstate { msg_store_clients = MSCState1 },
+                   {_SeqId, State3} = publish(Msg, MsgPropsFun(MsgProps),
+                                              true, true, State2),
+                   State3
+           end,
+           AckTags, State))).
 
 len(#vqstate { len = Len }) -> Len.
 
@@ -1156,7 +1153,6 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
                                    durable = IsDurable }) ->
     PAcks = lists:append(SPAcks),
     Acks  = lists:append(SAcks),
-    {_Guids, NewState} = ack(Acks, State),
     Pubs  = [{Msg, Fun(MsgProps)} || {Fun, PubsN}    <- lists:reverse(SPubs),
                                      {Msg, MsgProps} <- lists:reverse(PubsN)],
     {SeqIds, State1 = #vqstate { index_state = IndexState }} =
@@ -1168,7 +1164,7 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
                   {SeqId, State3} =
                       publish(Msg, MsgProps, false, IsPersistent1, State2),
                   {cons_if(IsPersistent1, SeqId, SeqIdsAcc), State3}
-          end, {PAcks, NewState}, Pubs),
+          end, {PAcks, ack(Acks, State)}, Pubs),
     IndexState1 = rabbit_queue_index:sync(SeqIds, IndexState),
     [ Fun() || Fun <- lists:reverse(SFuns) ],
     reduce_memory_use(
@@ -1319,7 +1315,7 @@ remove_pending_ack(KeepPersistent,
                    State = #vqstate { pending_ack       = PA,
                                       index_state       = IndexState,
                                       msg_store_clients = MSCState }) ->
-    {PersistentSeqIds, GuidsByStore, _AllGuids} =
+    {PersistentSeqIds, GuidsByStore} =
         dict:fold(fun accumulate_ack/3, accumulate_ack_init(), PA),
     State1 = State #vqstate { pending_ack   = dict:new(),
                               ram_ack_index = gb_trees:empty() },
@@ -1338,9 +1334,9 @@ remove_pending_ack(KeepPersistent,
     end.
 
 ack(_MsgStoreFun, _Fun, [], State) ->
-    {[], State};
+    State;
 ack(MsgStoreFun, Fun, AckTags, State) ->
-    {{PersistentSeqIds, GuidsByStore, AllGuids},
+    {{PersistentSeqIds, GuidsByStore},
      State1 = #vqstate { index_state       = IndexState,
                          msg_store_clients = MSCState,
                          persistent_count  = PCount,
@@ -1360,24 +1356,21 @@ ack(MsgStoreFun, Fun, AckTags, State) ->
      || {IsPersistent, Guids} <- orddict:to_list(GuidsByStore)],
     PCount1 = PCount - find_persistent_count(sum_guids_by_store_to_len(
                                                orddict:new(), GuidsByStore)),
-    {lists:reverse(AllGuids),
-     State1 #vqstate { index_state      = IndexState1,
-                       persistent_count = PCount1,
-                       ack_out_counter  = AckOutCount + length(AckTags) }}.
+    State1 #vqstate { index_state      = IndexState1,
+                      persistent_count = PCount1,
+                      ack_out_counter  = AckOutCount + length(AckTags) }.
 
-accumulate_ack_init() -> {[], orddict:new(), []}.
+accumulate_ack_init() -> {[], orddict:new()}.
 
 accumulate_ack(_SeqId, #msg_status { is_persistent = false, %% ASSERTIONS
                                      msg_on_disk   = false,
-                                     index_on_disk = false,
-                                     guid          = Guid },
-              {PersistentSeqIdsAcc, GuidsByStore, AllGuids}) ->
-    {PersistentSeqIdsAcc, GuidsByStore, [Guid | AllGuids]};
+                                     index_on_disk = false },
+              {PersistentSeqIdsAcc, GuidsByStore}) ->
+    {PersistentSeqIdsAcc, GuidsByStore};
 accumulate_ack(SeqId, {IsPersistent, Guid, _MsgProps},
-               {PersistentSeqIdsAcc, GuidsByStore, AllGuids}) ->
+               {PersistentSeqIdsAcc, GuidsByStore}) ->
     {cons_if(IsPersistent, SeqId, PersistentSeqIdsAcc),
-     rabbit_misc:orddict_cons(IsPersistent, Guid, GuidsByStore),
-     [Guid | AllGuids]}.
+     rabbit_misc:orddict_cons(IsPersistent, Guid, GuidsByStore)}.
 
 find_persistent_count(LensByStore) ->
     case orddict:find(true, LensByStore) of