Pretty much all the low hanging fruit. Need to check the ets access in txns in disk_queue and also deal with the clean shutdown and delivery bits.

> ** queue_prefetcher
> - s/publish/deliver ?

No, I really don't like that. Publish is about pushing messages to the receiver. Thus it's named correctly.

6 files changed, 189 insertions, 218 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 1d9f8c53cd..6c4c0ebb69 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -42,7 +42,7 @@
 -export([notify_sent/2, unblock/2]).
 -export([commit_all/2, rollback_all/2, notify_down_all/2, limit_all/3]).
 -export([on_node_down/1]).
--export([set_mode/2, report_memory/1]).
+-export([set_mode/2]).
 
 -import(mnesia).
 -import(gen_server2).
@@ -107,7 +107,6 @@
 -spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()).
 -spec(on_node_down/1 :: (erlang_node()) -> 'ok').
 -spec(pseudo_queue/2 :: (binary(), pid()) -> amqqueue()).
--spec(report_memory/1 :: (pid()) -> 'ok').
 
 -endif.
 
@@ -227,9 +226,6 @@ map(VHostPath, F) -> rabbit_misc:filter_exit_map(F, list(VHostPath)).
 set_mode(QPid, Mode) ->
     gen_server2:pcast(QPid, 10, {set_mode, Mode}).
 
-report_memory(QPid) ->
-    gen_server2:cast(QPid, report_memory).
-
 info(#amqqueue{ pid = QPid }) ->
     gen_server2:pcall(QPid, 9, info, infinity).
 
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 14a0370d83..b1c409b1f2 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -38,7 +38,7 @@
 -define(UNSENT_MESSAGE_LIMIT, 100).
 -define(HIBERNATE_AFTER_MIN, 1000).
 -define(DESIRED_HIBERNATE, 10000).
--define(MEMORY_REPORT_TIME_INTERVAL, 10000). %% 10 seconds in milliseconds
+-define(MINIMUM_MEMORY_REPORT_TIME_INTERVAL, 10000). %% 10 seconds in milliseconds
 
 -export([start_link/1]).
 
@@ -142,8 +142,8 @@ noreply(NewState) ->
     {noreply, start_memory_timer(NewState), hibernate}.
 
 start_memory_timer(State = #q { memory_report_timer = undefined }) ->
-    {ok, TRef} = timer:apply_after(?MEMORY_REPORT_TIME_INTERVAL,
-                                   rabbit_amqqueue, report_memory, [self()]),
+    {ok, TRef} = timer:send_after(?MINIMUM_MEMORY_REPORT_TIME_INTERVAL,
+                                  report_memory),
     report_memory(false, State #q { memory_report_timer = TRef });
 start_memory_timer(State) ->
     State.
@@ -199,11 +199,12 @@ record_current_channel_tx(ChPid, Txn) ->
     %% that wasn't happening already)
     store_ch_record((ch_record(ChPid))#cr{txn = Txn}).
     
-deliver_queue(Funs = {PredFun, DeliverFun}, FunAcc,
-              State = #q{q = #amqqueue{name = QName},
-                         active_consumers = ActiveConsumers,
-                         blocked_consumers = BlockedConsumers,
-                         next_msg_id = NextId}) ->
+deliver_msgs_to_consumers(
+  Funs = {PredFun, DeliverFun}, FunAcc,
+  State = #q{q = #amqqueue{name = QName},
+             active_consumers = ActiveConsumers,
+             blocked_consumers = BlockedConsumers,
+             next_msg_id = NextId}) ->
     case queue:out(ActiveConsumers) of
         {{value, QEntry = {ChPid, #consumer{tag = ConsumerTag,
                                             ack_required = AckRequired}}},
@@ -246,7 +247,7 @@ deliver_queue(Funs = {PredFun, DeliverFun}, FunAcc,
                                blocked_consumers = NewBlockedConsumers,
                                next_msg_id = NextId + 1
                                        },
-                    deliver_queue(Funs, FunAcc1, State2);
+                    deliver_msgs_to_consumers(Funs, FunAcc1, State2);
                 %% if IsMsgReady then we've hit the limiter
                 false when IsMsgReady ->
                     store_ch_record(C#cr{is_limit_active = true}),
@@ -254,7 +255,7 @@ deliver_queue(Funs = {PredFun, DeliverFun}, FunAcc,
                         move_consumers(ChPid,
                                        ActiveConsumers,
                                        BlockedConsumers),
-                    deliver_queue(
+                    deliver_msgs_to_consumers(
                       Funs, FunAcc,
                       State#q{active_consumers = NewActiveConsumers,
                               blocked_consumers = NewBlockedConsumers});
@@ -271,7 +272,7 @@ deliver_from_queue_pred({IsEmpty, _AutoAcks}, _State) ->
 deliver_from_queue_deliver(AckRequired, {false, AutoAcks},
                            State = #q { mixed_state = MS }) ->
     {{Msg, IsDelivered, AckTag, Remaining}, MS1} =
-        rabbit_mixed_queue:deliver(MS),
+        rabbit_mixed_queue:fetch(MS),
     AutoAcks1 =
         case AckRequired of
             true -> AutoAcks;
@@ -285,7 +286,7 @@ run_message_queue(State = #q { mixed_state = MS }) ->
              fun deliver_from_queue_deliver/3 },
     IsEmpty = rabbit_mixed_queue:is_empty(MS),
     {{_IsEmpty1, AutoAcks}, State1} =
-        deliver_queue(Funs, {IsEmpty, []}, State),
+        deliver_msgs_to_consumers(Funs, {IsEmpty, []}, State),
     {ok, MS1} =
         rabbit_mixed_queue:ack(AutoAcks, State1 #q.mixed_state),
     State1 #q { mixed_state = MS1 }.
@@ -306,7 +307,7 @@ attempt_immediate_delivery(none, _ChPid, Msg, State) ->
                     end,
                 {{Msg, false, AckTag}, true, State2}
         end,
-    deliver_queue({ PredFun, DeliverFun }, false, State);
+    deliver_msgs_to_consumers({ PredFun, DeliverFun }, false, State);
 attempt_immediate_delivery(Txn, ChPid, Msg, State) ->
     {ok, MS} = rabbit_mixed_queue:tx_publish(Msg, State #q.mixed_state),
     record_pending_message(Txn, ChPid, Msg),
@@ -330,8 +331,8 @@ deliver_or_requeue_n(MsgsWithAcks, State) ->
     Funs = { fun deliver_or_requeue_msgs_pred/2,
              fun deliver_or_requeue_msgs_deliver/3 },
     {{_RemainingLengthMinusOne, AutoAcks, OutstandingMsgs}, NewState} =
-        deliver_queue(Funs, {length(MsgsWithAcks) - 1, [], MsgsWithAcks},
-                      State),
+        deliver_msgs_to_consumers(
+          Funs, {length(MsgsWithAcks), [], MsgsWithAcks}, State),
     {ok, MS} = rabbit_mixed_queue:ack(AutoAcks,
                                       NewState #q.mixed_state),
     case OutstandingMsgs of
@@ -341,7 +342,7 @@ deliver_or_requeue_n(MsgsWithAcks, State) ->
     end.
 
 deliver_or_requeue_msgs_pred({Len, _AcksAcc, _MsgsWithAcks}, _State) ->
-    -1 < Len.
+    0 < Len.
 deliver_or_requeue_msgs_deliver(
   false, {Len, AcksAcc, [(MsgAckTag = {Msg, _}) | MsgsWithAcks]}, State) ->
     {{Msg, true, noack}, {Len - 1, [MsgAckTag | AcksAcc], MsgsWithAcks}, State};
@@ -612,11 +613,11 @@ handle_call({basic_get, ChPid, NoAck}, _From,
                        next_msg_id = NextId,
                        mixed_state = MS
                        }) ->
-    case rabbit_mixed_queue:deliver(MS) of
+    case rabbit_mixed_queue:fetch(MS) of
         {empty, MS1} -> reply(empty, State #q { mixed_state = MS1 });
         {{Msg, IsDelivered, AckTag, Remaining}, MS1} ->
             AckRequired = not(NoAck),
-            {ok, MS3} =
+            {ok, MS2} =
                 case AckRequired of
                     true ->
                         C = #cr{unacked_messages = UAM} = ch_record(ChPid),
@@ -628,9 +629,7 @@ handle_call({basic_get, ChPid, NoAck}, _From,
                 end,
             Message = {QName, self(), NextId, IsDelivered, Msg},
             reply({ok, Remaining, Message},
-                  State #q { next_msg_id = NextId + 1,
-                             mixed_state = MS3
-                           })
+                  State #q { next_msg_id = NextId + 1, mixed_state = MS2 })
     end;
 
 handle_call({basic_consume, NoAck, ReaderPid, ChPid, LimiterPid,
@@ -769,9 +768,9 @@ handle_cast({ack, Txn, MsgIds, ChPid}, State) ->
         not_found ->
             noreply(State);
         C = #cr{unacked_messages = UAM} ->
-            {MsgWithAcks, Remaining} = collect_messages(MsgIds, UAM),
             case Txn of
                 none ->
+                    {MsgWithAcks, Remaining} = collect_messages(MsgIds, UAM),
                     {ok, MS} =
                         rabbit_mixed_queue:ack(MsgWithAcks, State #q.mixed_state),
                     store_ch_record(C#cr{unacked_messages = Remaining}),
@@ -829,16 +828,13 @@ handle_cast({set_mode, Mode}, State = #q { mixed_state = MS }) ->
     PendingMessages =
         lists:flatten([Pending || #tx { pending_messages = Pending}
                                       <- all_tx_record()]),
-    {ok, MS1} = (case Mode of
-                    disk  -> fun rabbit_mixed_queue:to_disk_only_mode/2;
-                    mixed -> fun rabbit_mixed_queue:to_mixed_mode/2
-                 end)(PendingMessages, MS),
-    noreply(State #q { mixed_state = MS1 });
-
-handle_cast(report_memory, State) ->
-    %% deliberately don't call noreply/2 as we don't want to restart the timer
-    %% by unsetting the timer, we force a report on the next normal message
-    {noreply, State #q { memory_report_timer = undefined }, hibernate}.
+    {ok, MS1} = rabbit_mixed_queue:set_mode(Mode, PendingMessages, MS),
+    noreply(State #q { mixed_state = MS1 }).
+
+handle_info(report_memory, State) ->
+    %% deliberately don't call noreply/2 as we don't want to restart the timer.
+    %% By unsetting the timer, we force a report on the next normal message
+    {noreply, State #q { memory_report_timer = undefined }, hibernate};
 
 handle_info({'DOWN', MonitorRef, process, DownPid, _Reason},
             State = #q{owner = {DownPid, MonitorRef}}) ->
diff --git a/src/rabbit_disk_queue.erl b/src/rabbit_disk_queue.erl
index 5940f5ad04..e2f341ffd4 100644
--- a/src/rabbit_disk_queue.erl
+++ b/src/rabbit_disk_queue.erl
@@ -39,7 +39,7 @@
          terminate/2, code_change/3]).
 -export([handle_pre_hibernate/1]).
 
--export([publish/3, deliver/1, phantom_deliver/1, ack/2,
+-export([publish/3, fetch/1, phantom_fetch/1, ack/2,
          tx_publish/1, tx_commit/3, tx_cancel/1,
          requeue/2, purge/1, delete_queue/1,
          delete_non_durable_queues/1, auto_ack_next_message/1,
@@ -48,27 +48,27 @@
 
 -export([filesync/0, cache_info/0]).
 
--export([stop/0, stop_and_obliterate/0, report_memory/0,
-         set_mode/1, to_disk_only_mode/0, to_ram_disk_mode/0]).
+-export([stop/0, stop_and_obliterate/0, set_mode/1, to_disk_only_mode/0,
+         to_ram_disk_mode/0]).
 
 -include("rabbit.hrl").
 
--define(WRITE_OK_SIZE_BITS,          8).
--define(WRITE_OK_TRANSIENT,        255).
--define(WRITE_OK_PERSISTENT,       254).
--define(INTEGER_SIZE_BYTES,          8).
--define(INTEGER_SIZE_BITS,           (8 * ?INTEGER_SIZE_BYTES)).
--define(MSG_LOC_NAME,                rabbit_disk_queue_msg_location).
--define(FILE_SUMMARY_ETS_NAME,       rabbit_disk_queue_file_summary).
--define(SEQUENCE_ETS_NAME,           rabbit_disk_queue_sequences).
--define(CACHE_ETS_NAME,              rabbit_disk_queue_cache).
--define(FILE_EXTENSION,              ".rdq").
--define(FILE_EXTENSION_TMP,          ".rdt").
--define(FILE_EXTENSION_DETS,         ".dets").
--define(FILE_PACKING_ADJUSTMENT,     (1 + (2* (?INTEGER_SIZE_BYTES)))).
--define(MEMORY_REPORT_TIME_INTERVAL, 10000). %% 10 seconds in milliseconds
--define(BATCH_SIZE,                  10000).
--define(CACHE_MAX_SIZE,              10485760).
+-define(WRITE_OK_SIZE_BITS,                  8).
+-define(WRITE_OK_TRANSIENT,                255).
+-define(WRITE_OK_PERSISTENT,               254).
+-define(INTEGER_SIZE_BYTES,                  8).
+-define(INTEGER_SIZE_BITS,                   (8 * ?INTEGER_SIZE_BYTES)).
+-define(MSG_LOC_NAME,                        rabbit_disk_queue_msg_location).
+-define(FILE_SUMMARY_ETS_NAME,               rabbit_disk_queue_file_summary).
+-define(SEQUENCE_ETS_NAME,                   rabbit_disk_queue_sequences).
+-define(CACHE_ETS_NAME,                      rabbit_disk_queue_cache).
+-define(FILE_EXTENSION,                      ".rdq").
+-define(FILE_EXTENSION_TMP,                  ".rdt").
+-define(FILE_EXTENSION_DETS,                 ".dets").
+-define(FILE_PACKING_ADJUSTMENT,             (1 + (2* (?INTEGER_SIZE_BYTES)))).
+-define(MINIMUM_MEMORY_REPORT_TIME_INTERVAL, 10000). %% 10 seconds in millisecs
+-define(BATCH_SIZE,                          10000).
+-define(CACHE_MAX_SIZE,                      10485760).
 
 -define(SERVER, ?MODULE).
 
@@ -94,11 +94,11 @@
          file_size_limit,         %% how big can our files get?
          read_file_handles,       %% file handles for reading (LRU)
          read_file_handles_limit, %% how many file handles can we open?
-         on_sync_txns,           %% list of commiters to run on sync (reversed)
+         on_sync_txns,            %% list of commiters to run on sync (reversed)
          commit_timer_ref,        %% TRef for our interval timer
          last_sync_offset,        %% current_offset at the last time we sync'd
          message_cache,           %% ets message cache
-         memory_report_timer,     %% TRef for the memory report timer
+         memory_report_timer_ref, %% TRef for the memory report timer
          wordsize,                %% bytes in a word on this platform
          mnesia_bytes_per_record, %% bytes per record in mnesia in ram_disk mode
          ets_bytes_per_record     %% bytes per record in msg_location_ets
@@ -253,10 +253,10 @@
 -spec(start_link/0 :: () ->
               ({'ok', pid()} | 'ignore' | {'error', any()})).
 -spec(publish/3 :: (queue_name(), message(), bool()) -> 'ok').
--spec(deliver/1 :: (queue_name()) ->
+-spec(fetch/1 :: (queue_name()) ->
              ('empty' | {message(), non_neg_integer(),
                          bool(), {msg_id(), seq_id()}, non_neg_integer()})).
--spec(phantom_deliver/1 :: (queue_name()) ->
+-spec(phantom_fetch/1 :: (queue_name()) ->
              ( 'empty' | {msg_id(), bool(), bool(), {msg_id(), seq_id()},
                           non_neg_integer()})).
 -spec(prefetch/1 :: (queue_name()) -> 'ok'). 
@@ -281,7 +281,6 @@
 -spec(to_ram_disk_mode/0 :: () -> 'ok').
 -spec(filesync/0 :: () -> 'ok').
 -spec(cache_info/0 :: () -> [{atom(), term()}]).
--spec(report_memory/0 :: () -> 'ok').
 -spec(set_mode/1 :: ('disk' | 'mixed') -> 'ok').
 
 -endif.
@@ -295,11 +294,11 @@ start_link() ->
 publish(Q, Message = #basic_message {}, IsDelivered) ->
     gen_server2:cast(?SERVER, {publish, Q, Message, IsDelivered}).
 
-deliver(Q) ->
-    gen_server2:call(?SERVER, {deliver, Q}, infinity).
+fetch(Q) ->
+    gen_server2:call(?SERVER, {fetch, Q}, infinity).
 
-phantom_deliver(Q) ->
-    gen_server2:call(?SERVER, {phantom_deliver, Q}, infinity).
+phantom_fetch(Q) ->
+    gen_server2:call(?SERVER, {phantom_fetch, Q}, infinity).
 
 prefetch(Q) ->
     gen_server2:pcast(?SERVER, -1, {prefetch, Q, self()}).
@@ -360,9 +359,6 @@ filesync() ->
 cache_info() ->
     gen_server2:call(?SERVER, cache_info, infinity).
 
-report_memory() ->
-    gen_server2:cast(?SERVER, report_memory).
-
 set_mode(Mode) ->
     gen_server2:pcast(?SERVER, 10, {set_mode, Mode}).
 
@@ -406,8 +402,6 @@ init([FileSizeLimit, ReadFileHandlesLimit]) ->
     %% seems to blow up if it is set private
     MsgLocationEts = ets:new(?MSG_LOC_NAME, [set, protected]),
 
-    TRef = start_memory_timer(),
-
     InitName = "0" ++ ?FILE_EXTENSION,
     State =
         #dqstate { msg_location_dets       = MsgLocationDets,
@@ -430,7 +424,7 @@ init([FileSizeLimit, ReadFileHandlesLimit]) ->
                    last_sync_offset        = 0,
                    message_cache           = ets:new(?CACHE_ETS_NAME,
                                                      [set, private]),
-                   memory_report_timer     = TRef,
+                   memory_report_timer_ref = undefined,
                    wordsize                = erlang:system_info(wordsize),
                    mnesia_bytes_per_record = undefined,
                    ets_bytes_per_record    = undefined
@@ -457,14 +451,14 @@ init([FileSizeLimit, ReadFileHandlesLimit]) ->
     %% ets_bytes_per_record otherwise.
     ok = rabbit_queue_mode_manager:report_memory(self(), 0, false),
     ok = report_memory(false, State2),
-    {ok, State2, hibernate, {backoff, ?HIBERNATE_AFTER_MIN,
-                             ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
+    {ok, start_memory_timer(State2), hibernate,
+     {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
-handle_call({deliver, Q}, _From, State) ->
-    {ok, Result, State1} = internal_deliver(Q, true, false, true, State),
+handle_call({fetch, Q}, _From, State) ->
+    {ok, Result, State1} = internal_fetch(Q, true, false, true, State),
     reply(Result, State1);
-handle_call({phantom_deliver, Q}, _From, State) ->
-    {ok, Result, State1} = internal_deliver(Q, false, false, true, State),
+handle_call({phantom_fetch, Q}, _From, State) ->
+    {ok, Result, State1} = internal_fetch(Q, false, false, true, State),
     reply(Result, State1);
 handle_call({tx_commit, Q, PubMsgIds, AckSeqIds}, From, State) ->
     State1 =
@@ -534,13 +528,8 @@ handle_cast({set_mode, Mode}, State) ->
                  disk -> fun to_disk_only_mode/1;
                  mixed -> fun to_ram_disk_mode/1
              end)(State));
-handle_cast(report_memory, State) ->
-    %% call noreply1/2, not noreply/1/2, as we don't want to restart the
-    %% memory_report_timer
-    %% by unsetting the timer, we force a report on the next normal message
-    noreply1(State #dqstate { memory_report_timer = undefined });
 handle_cast({prefetch, Q, From}, State) ->
-    {ok, Result, State1} = internal_deliver(Q, true, true, false, State),
+    {ok, Result, State1} = internal_fetch(Q, true, true, false, State),
     Cont = rabbit_misc:with_exit_handler(
              fun () -> false end,
              fun () ->
@@ -550,7 +539,7 @@ handle_cast({prefetch, Q, From}, State) ->
     State3 =
 	case Cont of
 	    true ->
-		case internal_deliver(Q, false, false, true, State1) of
+		case internal_fetch(Q, false, false, true, State1) of
 		    {ok, empty, State2} -> State2;
 		    {ok, {_MsgId, _IsPersistent, _Delivered, _MsgSeqId, _Rem},
 		     State2} -> State2
@@ -559,6 +548,11 @@ handle_cast({prefetch, Q, From}, State) ->
 	end,
     noreply(State3).
         
+handle_info(report_memory, State) ->
+    %% call noreply1/2, not noreply/1/2, as we don't want to restart the
+    %% memory_report_timer_ref.
+    %% By unsetting the timer, we force a report on the next normal message
+    noreply1(State #dqstate { memory_report_timer_ref = undefined });
 handle_info({'EXIT', _Pid, Reason}, State) ->
     {stop, Reason, State};
 handle_info(timeout, State) ->
@@ -595,7 +589,7 @@ shutdown(State = #dqstate { msg_location_dets = MsgLocationDets,
     State1 #dqstate { current_file_handle = undefined,
                       current_dirty = false,
                       read_file_handles = {dict:new(), gb_trees:empty()},
-                      memory_report_timer = undefined
+                      memory_report_timer_ref = undefined
                     }.
 
 code_change(_OldVsn, State, _Extra) ->
@@ -603,20 +597,17 @@ code_change(_OldVsn, State, _Extra) ->
 
 %% ---- UTILITY FUNCTIONS ----
 
-stop_memory_timer(State = #dqstate { memory_report_timer = undefined }) ->
+stop_memory_timer(State = #dqstate { memory_report_timer_ref = undefined }) ->
     State;
-stop_memory_timer(State = #dqstate { memory_report_timer = TRef }) ->
+stop_memory_timer(State = #dqstate { memory_report_timer_ref = TRef }) ->
     {ok, cancel} = timer:cancel(TRef),
-    State #dqstate { memory_report_timer = undefined }.
-
-start_memory_timer() ->
-    {ok, TRef} = timer:apply_after(?MEMORY_REPORT_TIME_INTERVAL,
-                                   rabbit_disk_queue, report_memory, []),
-    TRef.
+    State #dqstate { memory_report_timer_ref = undefined }.
 
-start_memory_timer(State = #dqstate { memory_report_timer = undefined }) ->
+start_memory_timer(State = #dqstate { memory_report_timer_ref = undefined }) ->
     ok = report_memory(false, State),
-    State #dqstate { memory_report_timer = start_memory_timer() };
+    {ok, TRef} = timer:send_after(?MINIMUM_MEMORY_REPORT_TIME_INTERVAL,
+                                  report_memory),
+    State #dqstate { memory_report_timer_ref = TRef };
 start_memory_timer(State) ->
     State.
 
@@ -893,7 +884,7 @@ cache_is_full(#dqstate { message_cache = Cache }) ->
 
 %% ---- INTERNAL RAW FUNCTIONS ----
 
-internal_deliver(Q, ReadMsg, FakeDeliver, Advance,
+internal_fetch(Q, ReadMsg, FakeDeliver, Advance,
                  State = #dqstate { sequences = Sequences }) ->
     case sequence_lookup(Sequences, Q) of
         {SeqId, SeqId} -> {ok, empty, State};
@@ -971,7 +962,7 @@ internal_read_message(Q, ReadSeqId, ReadMsg, FakeDeliver, ForceInCache, State) -
     end.
 
 internal_auto_ack(Q, State) ->
-    case internal_deliver(Q, false, false, true, State) of
+    case internal_fetch(Q, false, false, true, State) of
         {ok, empty, State1} -> {ok, State1};
         {ok, {_MsgId, _IsPersistent, _Delivered, MsgSeqId, _Remaining},
          State1} ->
diff --git a/src/rabbit_mixed_queue.erl b/src/rabbit_mixed_queue.erl
index 3d989662db..2b25ab0fac 100644
--- a/src/rabbit_mixed_queue.erl
+++ b/src/rabbit_mixed_queue.erl
@@ -35,11 +35,11 @@
 
 -export([init/2]).
 
--export([publish/2, publish_delivered/2, deliver/1, ack/2,
+-export([publish/2, publish_delivered/2, fetch/1, ack/2,
          tx_publish/2, tx_commit/3, tx_cancel/2, requeue/2, purge/1,
          length/1, is_empty/1, delete_queue/1, maybe_prefetch/1]).
 
--export([to_disk_only_mode/2, to_mixed_mode/2, info/1,
+-export([set_mode/3, info/1,
          estimate_queue_memory_and_reset_counters/1]).
 
 -record(mqstate, { mode,
@@ -76,7 +76,7 @@
 -spec(publish/2 :: (message(), mqstate()) -> okmqs()).
 -spec(publish_delivered/2 :: (message(), mqstate()) ->
              {'ok', acktag(), mqstate()}).
--spec(deliver/1 :: (mqstate()) ->
+-spec(fetch/1 :: (mqstate()) ->
              {('empty' | {message(), boolean(), acktag(), non_neg_integer()}),
               mqstate()}).
 -spec(ack/2 :: ([{message(), acktag()}], mqstate()) -> okmqs()).
@@ -91,8 +91,7 @@
 -spec(length/1 :: (mqstate()) -> non_neg_integer()).
 -spec(is_empty/1 :: (mqstate()) -> boolean()).
 
--spec(to_disk_only_mode/2 :: ([message()], mqstate()) -> okmqs()).
--spec(to_mixed_mode/2 :: ([message()], mqstate()) -> okmqs()).
+-spec(set_mode/3 :: (mode(), [message()], mqstate()) -> okmqs()).
 
 -spec(estimate_queue_memory_and_reset_counters/1 :: (mqstate()) ->
              {mqstate(), non_neg_integer(), non_neg_integer(),
@@ -120,8 +119,13 @@ size_of_message(
                         SumAcc + size(Frag)
                 end, 0, Payload).
 
-to_disk_only_mode(_TxnMessages, State = #mqstate { mode = disk }) ->
+set_mode(Mode, _TxnMessages, State = #mqstate { mode = Mode }) ->
     {ok, State};
+set_mode(disk, TxnMessages, State) ->
+    to_disk_only_mode(TxnMessages, State);
+set_mode(mixed, TxnMessages, State) ->
+    to_mixed_mode(TxnMessages, State).
+
 to_disk_only_mode(TxnMessages, State =
                   #mqstate { mode = mixed, queue = Q, msg_buf = MsgBuf,
                              is_durable = IsDurable, prefetcher = Prefetcher
@@ -219,8 +223,6 @@ flush_requeue_to_disk_queue(Q, RequeueCount, Commit) ->
     ok = rabbit_disk_queue:requeue_next_n(Q, RequeueCount),
     [].
 
-to_mixed_mode(_TxnMessages, State = #mqstate { mode = mixed }) ->
-    {ok, State};
 to_mixed_mode(TxnMessages, State = #mqstate { mode = disk, queue = Q,
                                               is_durable = IsDurable }) ->
     rabbit_log:info("Converting queue to mixed mode: ~p~n", [Q]),
@@ -248,6 +250,16 @@ to_mixed_mode(TxnMessages, State = #mqstate { mode = disk, queue = Q,
     garbage_collect(),
     {ok, State #mqstate { mode = mixed }}.
 
+gain_memory(Inc, State = #mqstate { memory_size = QSize,
+                                    memory_gain = Gain }) ->
+    State #mqstate { memory_size = QSize + Inc,
+                     memory_gain = Gain + Inc }.
+
+lose_memory(Dec, State = #mqstate { memory_size = QSize,
+                                    memory_loss = Loss }) ->
+    State #mqstate { memory_size = QSize - Dec,
+                     memory_loss = Loss + Dec }.
+
 inc_queue_length(_Q, MsgBuf, 0) ->
     MsgBuf;
 inc_queue_length(Q, MsgBuf, Count) ->
@@ -264,7 +276,7 @@ dec_queue_length(Count, State = #mqstate { queue = Q, msg_buf = MsgBuf }) ->
         {{value, {Q, Len}}, MsgBuf1} ->
             case Len of
                 Count ->
-                    maybe_prefetch(State #mqstate { msg_buf = MsgBuf1 });
+                    State #mqstate { msg_buf = MsgBuf1 };
                 _ when Len > Count ->
                     State #mqstate { msg_buf = queue:in_r({Q, Len-Count},
                                                           MsgBuf1)}
@@ -286,26 +298,23 @@ maybe_prefetch(State) ->
     State.
 
 publish(Msg, State = #mqstate { mode = disk, queue = Q, length = Length,
-                                msg_buf = MsgBuf, memory_size = QSize,
-                                memory_gain = Gain }) ->
+                                msg_buf = MsgBuf }) ->
     MsgBuf1 = inc_queue_length(Q, MsgBuf, 1),
     ok = rabbit_disk_queue:publish(Q, Msg, false),
     MsgSize = size_of_message(Msg),
-    {ok, State #mqstate { memory_gain = Gain + MsgSize,
-                          memory_size = QSize + MsgSize,
-                          msg_buf = MsgBuf1, length = Length + 1 }};
+    {ok, gain_memory(MsgSize, State #mqstate { msg_buf = MsgBuf1,
+                                               length = Length + 1 })};
 publish(Msg = #basic_message { is_persistent = IsPersistent }, State = 
         #mqstate { queue = Q, mode = mixed, is_durable = IsDurable,
-                   msg_buf = MsgBuf, length = Length, memory_size = QSize,
-                   memory_gain = Gain }) ->
+                   msg_buf = MsgBuf, length = Length }) ->
     ok = case IsDurable andalso IsPersistent of
              true -> rabbit_disk_queue:publish(Q, Msg, false);
              false -> ok
          end,
     MsgSize = size_of_message(Msg),
-    {ok, State #mqstate { msg_buf = queue:in({Msg, false}, MsgBuf),
-                          length = Length + 1, memory_size = QSize + MsgSize,
-                          memory_gain = Gain + MsgSize }}.
+    {ok, gain_memory(MsgSize,
+                     State #mqstate { msg_buf = queue:in({Msg, false}, MsgBuf),
+                                      length = Length + 1 })}.
 
 %% Assumption here is that the queue is empty already (only called via
 %% attempt_immediate_delivery).
@@ -313,20 +322,18 @@ publish_delivered(Msg =
                   #basic_message { guid = MsgId, is_persistent = IsPersistent},
                   State =
                   #mqstate { mode = Mode, is_durable = IsDurable,
-                             queue = Q, length = 0,
-                             memory_size = QSize, memory_gain = Gain })
+                             queue = Q, length = 0 })
   when Mode =:= disk orelse (IsDurable andalso IsPersistent) ->
     ok = rabbit_disk_queue:publish(Q, Msg, true),
     MsgSize = size_of_message(Msg),
-    State1 = State #mqstate { memory_size = QSize + MsgSize,
-                              memory_gain = Gain + MsgSize },
+    State1 = gain_memory(MsgSize, State),
     case IsDurable andalso IsPersistent of
         true ->
             %% must call phantom_deliver otherwise the msg remains at
             %% the head of the queue. This is synchronous, but
             %% unavoidable as we need the AckTag
             {MsgId, IsPersistent, true, AckTag, 0} =
-                rabbit_disk_queue:phantom_deliver(Q),
+                rabbit_disk_queue:phantom_fetch(Q),
             {ok, AckTag, State1};
         false ->
             %% in this case, we don't actually care about the ack, so
@@ -334,18 +341,15 @@ publish_delivered(Msg =
             ok = rabbit_disk_queue:auto_ack_next_message(Q),
             {ok, noack, State1}
     end;
-publish_delivered(Msg, State =
-                  #mqstate { mode = mixed, length = 0, memory_size = QSize,
-                             memory_gain = Gain }) ->
+publish_delivered(Msg, State = #mqstate { mode = mixed, length = 0 }) ->
     MsgSize = size_of_message(Msg),
-    {ok, noack, State #mqstate { memory_size = QSize + MsgSize,
-                                 memory_gain = Gain + MsgSize }}.
+    {ok, noack, gain_memory(MsgSize, State)}.
 
-deliver(State = #mqstate { length = 0 }) ->
+fetch(State = #mqstate { length = 0 }) ->
     {empty, State};
-deliver(State = #mqstate { msg_buf = MsgBuf, queue = Q,
-                           is_durable = IsDurable, length = Length,
-                           prefetcher = Prefetcher }) ->
+fetch(State = #mqstate { msg_buf = MsgBuf, queue = Q,
+                         is_durable = IsDurable, length = Length,
+                         prefetcher = Prefetcher }) ->
     {{value, Value}, MsgBuf1} = queue:out(MsgBuf),
     Rem = Length - 1,
     State1 = State #mqstate { length = Rem },
@@ -356,13 +360,13 @@ deliver(State = #mqstate { msg_buf = MsgBuf, queue = Q,
                 case IsDurable andalso IsPersistent of
                     true ->
                         {MsgId, IsPersistent, IsDelivered, AckTag1, _PRem}
-                            = rabbit_disk_queue:phantom_deliver(Q),
+                            = rabbit_disk_queue:phantom_fetch(Q),
                         AckTag1;
                     false ->
                         noack
                 end,
-            State2 = maybe_prefetch(State1 #mqstate { msg_buf = MsgBuf1 }),
-            {{Msg, IsDelivered, AckTag, Rem}, State2};
+            {{Msg, IsDelivered, AckTag, Rem},
+             State1 #mqstate { msg_buf = MsgBuf1 }};
         {Msg = #basic_message { is_persistent = IsPersistent },
          IsDelivered, AckTag} ->
             %% message has come via the prefetcher, thus it's been
@@ -375,21 +379,21 @@ deliver(State = #mqstate { msg_buf = MsgBuf, queue = Q,
             State2 = dec_queue_length(1, State1),
             {Msg = #basic_message { is_persistent = IsPersistent },
              _Size, IsDelivered, AckTag, _PersistRem}
-                = rabbit_disk_queue:deliver(Q),
+                = rabbit_disk_queue:fetch(Q),
             AckTag1 = maybe_ack(Q, IsDurable, IsPersistent, AckTag),
             {{Msg, IsDelivered, AckTag1, Rem}, State2};
         _ ->
             case rabbit_queue_prefetcher:drain(Prefetcher) of
-                empty -> deliver(State #mqstate { prefetcher = undefined });
+                empty -> fetch(State #mqstate { prefetcher = undefined });
                 {Fetched, Len, Status} ->
                     State2 = #mqstate { msg_buf = MsgBuf2 } =
                         dec_queue_length(Len, State),
-                    deliver(State2 #mqstate
-                            { msg_buf = queue:join(Fetched, MsgBuf2),
-                              prefetcher = case Status of
-                                               finished -> undefined;
-                                               continuing -> Prefetcher
-                                           end })
+                    fetch(State2 #mqstate
+                          { msg_buf = queue:join(Fetched, MsgBuf2),
+                            prefetcher = case Status of
+                                             finished -> undefined;
+                                             continuing -> Prefetcher
+                                         end })
             end
     end.
 
@@ -407,38 +411,30 @@ remove_noacks(MsgsWithAcks) ->
               {[AckTag | AccAckTags], size_of_message(Msg) + AccSize}
       end, {[], 0}, MsgsWithAcks).
 
-ack(MsgsWithAcks, State = #mqstate { queue = Q, memory_size = QSize,
-                                     memory_loss = Loss }) ->
+ack(MsgsWithAcks, State = #mqstate { queue = Q }) ->
     {AckTags, ASize} = remove_noacks(MsgsWithAcks),
     ok = case AckTags of
              [] -> ok;
              _ -> rabbit_disk_queue:ack(Q, AckTags)
          end,
-    State1 = State #mqstate { memory_size = QSize - ASize,
-                              memory_loss = Loss + ASize },
-    {ok, State1}.
+    {ok, lose_memory(ASize, State)}.
                                                    
 tx_publish(Msg = #basic_message { is_persistent = IsPersistent },
-           State = #mqstate { mode = Mode, memory_size = QSize,
-                              is_durable = IsDurable, memory_gain = Gain })
+           State = #mqstate { mode = Mode, is_durable = IsDurable })
   when Mode =:= disk orelse (IsDurable andalso IsPersistent) ->
     ok = rabbit_disk_queue:tx_publish(Msg),
     MsgSize = size_of_message(Msg),
-    {ok, State #mqstate { memory_size = QSize + MsgSize,
-                          memory_gain = Gain + MsgSize }};
-tx_publish(Msg, State = #mqstate { mode = mixed, memory_size = QSize,
-                                   memory_gain = Gain }) ->
+    {ok, gain_memory(MsgSize, State)};
+tx_publish(Msg, State = #mqstate { mode = mixed }) ->
     %% this message will reappear in the tx_commit, so ignore for now
     MsgSize = size_of_message(Msg),
-    {ok, State #mqstate { memory_size = QSize + MsgSize,
-                          memory_gain = Gain + MsgSize }}.
+    {ok, gain_memory(MsgSize, State)}.
 
 only_msg_ids(Pubs) ->
     lists:map(fun (Msg) -> {Msg #basic_message.guid, false} end, Pubs).
 
 tx_commit(Publishes, MsgsWithAcks,
           State = #mqstate { mode = disk, queue = Q, length = Length,
-                             memory_size = QSize, memory_loss = Loss,
                              msg_buf = MsgBuf }) ->
     {RealAcks, ASize} = remove_noacks(MsgsWithAcks),
     ok = if ([] == Publishes) andalso ([] == RealAcks) -> ok;
@@ -446,14 +442,12 @@ tx_commit(Publishes, MsgsWithAcks,
                                                 RealAcks)
          end,
     Len = erlang:length(Publishes),
-    {ok, State #mqstate { length = Length + Len,
-                          msg_buf = inc_queue_length(Q, MsgBuf, Len),
-                          memory_size = QSize - ASize,
-                          memory_loss = Loss + ASize }};
+    {ok, lose_memory(ASize, State #mqstate
+                     { length = Length + Len,
+                       msg_buf = inc_queue_length(Q, MsgBuf, Len) })};
 tx_commit(Publishes, MsgsWithAcks,
           State = #mqstate { mode = mixed, queue = Q, msg_buf = MsgBuf,
-                             is_durable = IsDurable, length = Length,
-                             memory_size = QSize, memory_loss = Loss }) ->
+                             is_durable = IsDurable, length = Length }) ->
     {PersistentPubs, MsgBuf1} =
         lists:foldl(fun (Msg = #basic_message { is_persistent = IsPersistent },
                          {Acc, MsgBuf2}) ->
@@ -471,23 +465,20 @@ tx_commit(Publishes, MsgsWithAcks,
              false -> rabbit_disk_queue:tx_commit(
                         Q, lists:reverse(PersistentPubs), RealAcks)
          end,
-    {ok, State #mqstate { msg_buf = MsgBuf1, memory_size = QSize - ASize,
-                          length = Length + erlang:length(Publishes),
-                          memory_loss = Loss + ASize }}.
+    {ok, lose_memory(ASize, State #mqstate
+                     { msg_buf = MsgBuf1,
+                       length = Length + erlang:length(Publishes) })}.
 
-tx_cancel(Publishes, State = #mqstate { mode = disk, memory_size = QSize,
-                                        memory_loss = Loss }) ->
+tx_cancel(Publishes, State = #mqstate { mode = disk }) ->
     {MsgIds, CSize} =
         lists:foldl(
           fun (Msg = #basic_message { guid = MsgId }, {MsgIdsAcc, CSizeAcc}) ->
                   {[MsgId | MsgIdsAcc], CSizeAcc + size_of_message(Msg)}
           end, {[], 0}, Publishes),
     ok = rabbit_disk_queue:tx_cancel(MsgIds),
-    {ok, State #mqstate { memory_size = QSize - CSize,
-                          memory_loss = Loss + CSize }};
-tx_cancel(Publishes, State = #mqstate { mode = mixed, is_durable = IsDurable,
-                                        memory_size = QSize,
-                                        memory_loss = Loss }) ->
+    {ok, lose_memory(CSize, State)};
+tx_cancel(Publishes,
+          State = #mqstate { mode = mixed, is_durable = IsDurable }) ->
     {PersistentPubs, CSize} =
         lists:foldl(
           fun (Msg = #basic_message { is_persistent = IsPersistent,
@@ -503,8 +494,7 @@ tx_cancel(Publishes, State = #mqstate { mode = mixed, is_durable = IsDurable,
                 rabbit_disk_queue:tx_cancel(PersistentPubs);
            true -> ok
         end,
-    {ok, State #mqstate { memory_size = QSize - CSize,
-                          memory_loss = Loss + CSize }}.
+    {ok, lose_memory(CSize, State)}.
 
 %% [{Msg, AckTag}]
 requeue(MessagesWithAckTags, State = #mqstate { mode = disk, queue = Q,
@@ -555,32 +545,30 @@ requeue(MessagesWithAckTags, State = #mqstate { mode = mixed, queue = Q,
                          length = Length + erlang:length(MessagesWithAckTags)}}.
 
 purge(State = #mqstate { queue = Q, mode = disk, length = Count,
-                         memory_loss = Loss, memory_size = QSize }) ->
+                         memory_size = QSize }) ->
     Count = rabbit_disk_queue:purge(Q),
-    {Count, State #mqstate { length = 0, memory_size = 0,
-                             memory_loss = Loss + QSize }};
+    {Count, lose_memory(QSize, State)};
 purge(State = #mqstate { queue = Q, mode = mixed, length = Length,
-                         memory_loss = Loss, memory_size = QSize,
-                         prefetcher = Prefetcher }) ->
+                         memory_size = QSize, prefetcher = Prefetcher }) ->
     case Prefetcher of
         undefined -> ok;
         _ -> rabbit_queue_prefetcher:drain_and_stop(Prefetcher)
     end,
     rabbit_disk_queue:purge(Q),
-    {Length,
-     State #mqstate { msg_buf = queue:new(), length = 0, memory_size = 0,
-                      memory_loss = Loss + QSize, prefetcher = undefined }}.
+    {Length, lose_memory(QSize, State #mqstate { msg_buf = queue:new(),
+                                                 length = 0,
+                                                 prefetcher = undefined })}.
 
 delete_queue(State = #mqstate { queue = Q, memory_size = QSize,
-                                memory_loss = Loss, prefetcher = Prefetcher
+                                prefetcher = Prefetcher
                               }) ->
     case Prefetcher of
         undefined -> ok;
         _ -> rabbit_queue_prefetcher:drain_and_stop(Prefetcher)
     end,
     ok = rabbit_disk_queue:delete_queue(Q),
-    {ok, State #mqstate { length = 0, memory_size = 0, msg_buf = queue:new(),
-                          memory_loss = Loss + QSize, prefetcher = undefined }}.
+    {ok, lose_memory(QSize, State #mqstate { length = 0, msg_buf = queue:new(),
+                                             prefetcher = undefined })}.
 
 length(#mqstate { length = Length }) ->
     Length.
diff --git a/src/rabbit_queue_prefetcher.erl b/src/rabbit_queue_prefetcher.erl
index c847848de8..ad6b1ce2d4 100644
--- a/src/rabbit_queue_prefetcher.erl
+++ b/src/rabbit_queue_prefetcher.erl
@@ -94,7 +94,7 @@
 %%    to its internal queue. A cast is not sufficient here because the
 %%    mixed_queue could come along, drain the prefetcher, thus
 %%    catching the msg just sent by the disk_queue and then call
-%%    disk_queue:deliver(Q) which is normal priority call, which could
+%%    disk_queue:fetch(Q) which is normal priority call, which could
 %%    overtake a reply cast from the prefetcher to the disk queue,
 %%    which would result in the same message being delivered
 %%    twice. Thus when the disk_queue calls prefetcher:publish(Msg),
@@ -146,7 +146,7 @@
 %% mixed_queue tries to drain the prefetcher. We must therefore ensure
 %% that this msg can't also be delivered to the mixed_queue directly
 %% by the disk_queue through the mixed_queue calling
-%% disk_queue:deliver(Q) which is why the prefetcher:publish function
+%% disk_queue:fetch(Q) which is why the prefetcher:publish function
 %% is a call and not a cast, thus blocking the disk_queue.
 %%
 %% Finally, the prefetcher is only created when the mixed_queue is
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index ffd675a0b7..ad5a248314 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -157,7 +157,7 @@ test_simple_n_element_queue(N) ->
     passed.
 
 test_unfold() ->
-    {[], test} = rabbit_misc:unfold(fun (V) -> false end, test),
+    {[], test} = rabbit_misc:unfold(fun (_V) -> false end, test),
     List = lists:seq(2,20,2),
     {List, 0} = rabbit_misc:unfold(fun (0) -> false;
                                        (N) -> {true, N*2, N-1}
@@ -848,7 +848,7 @@ rdq_time_tx_publish_commit_deliver_ack(Qs, MsgCount, MsgSizeBytes) ->
                                 [begin
                                      Remaining = MsgCount - N,
                                      {Message, _TSize, false, SeqId,
-                                      Remaining} = rabbit_disk_queue:deliver(Q),
+                                      Remaining} = rabbit_disk_queue:fetch(Q),
                                      ok = rdq_match_message(Message, N, Msg, MsgSizeBytes),
                                      SeqId
                                  end || N <- List],
@@ -895,7 +895,7 @@ rdq_stress_gc(MsgCount) ->
           fun (MsgId, Acc) ->
                   Remaining = MsgCount - MsgId,
                   {Message, _TSize, false, SeqId, Remaining} =
-                      rabbit_disk_queue:deliver(q),
+                      rabbit_disk_queue:fetch(q),
                   ok = rdq_match_message(Message, MsgId, Msg, MsgSizeBytes),
                   dict:store(MsgId, SeqId, Acc)
           end, dict:new(), List),
@@ -904,7 +904,7 @@ rdq_stress_gc(MsgCount) ->
            rabbit_disk_queue:ack(q, [SeqId])
      end || MsgId <- AckList2],
     rabbit_disk_queue:tx_commit(q, [], []),
-    empty = rabbit_disk_queue:deliver(q),
+    empty = rabbit_disk_queue:fetch(q),
     rdq_stop(),
     passed.
 
@@ -923,7 +923,7 @@ rdq_test_startup_with_queue_gaps() ->
     Seqs = [begin
                 Remaining = Total - N,
                 {Message, _TSize, false, SeqId, Remaining} =
-                    rabbit_disk_queue:deliver(q),
+                    rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
             end || N <- lists:seq(1,Half)],
@@ -945,7 +945,7 @@ rdq_test_startup_with_queue_gaps() ->
     Seqs2 = [begin
                  Remaining = round(Total - ((Half + N)/2)),
                  {Message, _TSize, true, SeqId, Remaining} =
-                     rabbit_disk_queue:deliver(q),
+                     rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
              end || N <- lists:seq(2,Half,2)],
@@ -955,13 +955,13 @@ rdq_test_startup_with_queue_gaps() ->
     Seqs3 = [begin
                  Remaining = Total - N,
                  {Message, _TSize, false, SeqId, Remaining} =
-                     rabbit_disk_queue:deliver(q),
+                     rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
              end || N <- lists:seq(1 + Half,Total)],
     rabbit_disk_queue:tx_commit(q, [], Seqs3),
     io:format("Read second half done~n", []),
-    empty = rabbit_disk_queue:deliver(q),
+    empty = rabbit_disk_queue:fetch(q),
     rdq_stop(),
     passed.
 
@@ -980,7 +980,7 @@ rdq_test_redeliver() ->
     Seqs = [begin
                 Remaining = Total - N,
                 {Message, _TSize, false, SeqId, Remaining} =
-                    rabbit_disk_queue:deliver(q),
+                    rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
             end || N <- lists:seq(1,Half)],
@@ -1001,7 +1001,7 @@ rdq_test_redeliver() ->
     Seqs2 = [begin
                  Remaining = round(Total - N + (Half/2)),
                  {Message, _TSize, false, SeqId, Remaining} =
-                     rabbit_disk_queue:deliver(q),
+                     rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
              end || N <- lists:seq(1+Half, Total)],
@@ -1009,12 +1009,12 @@ rdq_test_redeliver() ->
     Seqs3 = [begin
                  Remaining = round((Half - N) / 2) - 1,
                  {Message, _TSize, true, SeqId, Remaining} =
-                     rabbit_disk_queue:deliver(q),
+                     rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
              end || N <- lists:seq(1, Half, 2)],
     rabbit_disk_queue:tx_commit(q, [], Seqs3),
-    empty = rabbit_disk_queue:deliver(q),
+    empty = rabbit_disk_queue:fetch(q),
     rdq_stop(),
     passed.
 
@@ -1033,7 +1033,7 @@ rdq_test_purge() ->
     Seqs = [begin
                 Remaining = Total - N,
                 {Message, _TSize, false, SeqId, Remaining} =
-                    rabbit_disk_queue:deliver(q),
+                    rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
             end || N <- lists:seq(1,Half)],
@@ -1042,7 +1042,7 @@ rdq_test_purge() ->
     io:format("Purge done~n", []),
     rabbit_disk_queue:tx_commit(q, [], Seqs),
     io:format("Ack first half done~n", []),
-    empty = rabbit_disk_queue:deliver(q),
+    empty = rabbit_disk_queue:fetch(q),
     rdq_stop(),
     passed.    
 
@@ -1051,7 +1051,7 @@ rdq_new_mixed_queue(Q, Durable, Disk) ->
     {MS1, _, _, _} =
         rabbit_mixed_queue:estimate_queue_memory_and_reset_counters(MS),
     case Disk of
-        true -> {ok, MS2} = rabbit_mixed_queue:to_disk_only_mode([], MS1),
+        true -> {ok, MS2} = rabbit_mixed_queue:set_mode(disk, [], MS1),
                 MS2;
         false -> MS1
     end.
@@ -1083,11 +1083,11 @@ rdq_test_mixed_queue_modes() ->
     30 = rabbit_mixed_queue:length(MS6),
     io:format("Published a mixture of messages; ~w~n",
               [rabbit_mixed_queue:estimate_queue_memory_and_reset_counters(MS6)]),
-    {ok, MS7} = rabbit_mixed_queue:to_disk_only_mode([], MS6),
+    {ok, MS7} = rabbit_mixed_queue:set_mode(disk, [], MS6),
     30 = rabbit_mixed_queue:length(MS7),
     io:format("Converted to disk only mode; ~w~n",
              [rabbit_mixed_queue:estimate_queue_memory_and_reset_counters(MS7)]),
-    {ok, MS8} = rabbit_mixed_queue:to_mixed_mode([], MS7),
+    {ok, MS8} = rabbit_mixed_queue:set_mode(mixed, [], MS7),
     30 = rabbit_mixed_queue:length(MS8),
     io:format("Converted to mixed mode; ~w~n",
               [rabbit_mixed_queue:estimate_queue_memory_and_reset_counters(MS8)]),
@@ -1097,12 +1097,12 @@ rdq_test_mixed_queue_modes() ->
                   Rem = 30 - N,
                   {{#basic_message { is_persistent = false },
                     false, _AckTag, Rem},
-                   MS9a} = rabbit_mixed_queue:deliver(MS9),
+                   MS9a} = rabbit_mixed_queue:fetch(MS9),
                   MS9a
           end, MS8, lists:seq(1,10)),
     20 = rabbit_mixed_queue:length(MS10),
     io:format("Delivered initial non persistent messages~n"),
-    {ok, MS11} = rabbit_mixed_queue:to_disk_only_mode([], MS10),
+    {ok, MS11} = rabbit_mixed_queue:set_mode(disk, [], MS10),
     20 = rabbit_mixed_queue:length(MS11),
     io:format("Converted to disk only mode~n"),
     rdq_stop(),
@@ -1116,13 +1116,13 @@ rdq_test_mixed_queue_modes() ->
                   Rem = 10 - N,
                   {{Msg = #basic_message { is_persistent = true },
                     false, AckTag, Rem},
-                   MS13a} = rabbit_mixed_queue:deliver(MS13),
+                   MS13a} = rabbit_mixed_queue:fetch(MS13),
                   {MS13a, [{Msg, AckTag} | AcksAcc]}
           end, {MS12, []}, lists:seq(1,10)),
     0 = rabbit_mixed_queue:length(MS14),
     {ok, MS15} = rabbit_mixed_queue:ack(AckTags, MS14),
     io:format("Delivered and acked all messages~n"),
-    {ok, MS16} = rabbit_mixed_queue:to_disk_only_mode([], MS15),
+    {ok, MS16} = rabbit_mixed_queue:set_mode(disk, [], MS15),
     0 = rabbit_mixed_queue:length(MS16),
     io:format("Converted to disk only mode~n"),
     rdq_stop(),
@@ -1149,28 +1149,28 @@ rdq_test_mode_conversion_mid_txn() ->
     rdq_start(),
     MS0 = rdq_new_mixed_queue(q, true, false),
     passed = rdq_tx_publish_mixed_alter_commit_get(
-               MS0, MsgsA, MsgsB, fun rabbit_mixed_queue:to_disk_only_mode/2, commit),
+               MS0, MsgsA, MsgsB, disk, commit),
 
     rdq_stop_virgin_start(),
     MS1 = rdq_new_mixed_queue(q, true, false),
     passed = rdq_tx_publish_mixed_alter_commit_get(
-               MS1, MsgsA, MsgsB, fun rabbit_mixed_queue:to_disk_only_mode/2, cancel),
+               MS1, MsgsA, MsgsB, disk, cancel),
 
 
     rdq_stop_virgin_start(),
     MS2 = rdq_new_mixed_queue(q, true, true),
     passed = rdq_tx_publish_mixed_alter_commit_get(
-               MS2, MsgsA, MsgsB, fun rabbit_mixed_queue:to_mixed_mode/2, commit),
+               MS2, MsgsA, MsgsB, mixed, commit),
 
     rdq_stop_virgin_start(),
     MS3 = rdq_new_mixed_queue(q, true, true),
     passed = rdq_tx_publish_mixed_alter_commit_get(
-               MS3, MsgsA, MsgsB, fun rabbit_mixed_queue:to_mixed_mode/2, cancel),
+               MS3, MsgsA, MsgsB, mixed, cancel),
 
     rdq_stop(),
     passed.
 
-rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, ChangeFun, CommitOrCancel) ->
+rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, Mode, CommitOrCancel) ->
     0 = rabbit_mixed_queue:length(MS0),
     MS2 = lists:foldl(
             fun (Msg, MS1) ->
@@ -1185,7 +1185,7 @@ rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, ChangeFun, CommitOrCanc
                     MS3a
             end, MS2, MsgsB),
     Len0 = rabbit_mixed_queue:length(MS4),
-    {ok, MS5} = ChangeFun(MsgsB, MS4),
+    {ok, MS5} = rabbit_mixed_queue:set_mode(Mode, MsgsB, MS4),
     Len0 = rabbit_mixed_queue:length(MS5),
     {ok, MS9} =
         case CommitOrCancel of
@@ -1198,7 +1198,7 @@ rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, ChangeFun, CommitOrCanc
                       fun (Msg, {Acc, MS7}) ->
                               Rem = Len1 - (Msg #basic_message.guid) - 1,
                               {{Msg, false, AckTag, Rem}, MS7a} =
-                                  rabbit_mixed_queue:deliver(MS7),
+                                  rabbit_mixed_queue:fetch(MS7),
                               {[{Msg, AckTag} | Acc], MS7a}
                       end, {[], MS6}, MsgsA ++ MsgsB),
                 0 = rabbit_mixed_queue:length(MS8),
@@ -1211,7 +1211,7 @@ rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, ChangeFun, CommitOrCanc
                       fun (Msg, {Acc, MS7}) ->
                               Rem = Len0 - (Msg #basic_message.guid) - 1,
                               {{Msg, false, AckTag, Rem}, MS7a} =
-                                  rabbit_mixed_queue:deliver(MS7),
+                                  rabbit_mixed_queue:fetch(MS7),
                               {[{Msg, AckTag} | Acc], MS7a}
                       end, {[], MS6}, MsgsA),
                 0 = rabbit_mixed_queue:length(MS8),
@@ -1244,7 +1244,7 @@ rdq_test_disk_queue_modes() ->
     Seqs = [begin
                 Remaining = Total - N,
                 {Message, _TSize, false, SeqId, Remaining} =
-                    rabbit_disk_queue:deliver(q),
+                    rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
             end || N <- Half1],
@@ -1254,7 +1254,7 @@ rdq_test_disk_queue_modes() ->
     Seqs2 = [begin
                  Remaining = Total - N,
                  {Message, _TSize, false, SeqId, Remaining} =
-                     rabbit_disk_queue:deliver(q),
+                     rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
              end || N <- Half2],
@@ -1262,7 +1262,7 @@ rdq_test_disk_queue_modes() ->
     ok = rabbit_disk_queue:tx_commit(q, [], Seqs),
     ok = rabbit_disk_queue:to_disk_only_mode(),
     ok = rabbit_disk_queue:tx_commit(q, [], Seqs2),
-    empty = rabbit_disk_queue:deliver(q),
+    empty = rabbit_disk_queue:fetch(q),
     rdq_stop(),
     passed.