typo in docs in previous commit.

Also changed internal_fetch so its result construction, which whilst not wrong, was at least confusing, and had unexecutable code in it. Associated changes elsewhere.

4 files changed, 29 insertions, 37 deletions

diff --git a/src/rabbit_disk_queue.erl b/src/rabbit_disk_queue.erl
index 96125031dc..cb2487f3d2 100644
--- a/src/rabbit_disk_queue.erl
+++ b/src/rabbit_disk_queue.erl
@@ -254,10 +254,10 @@
               ({'ok', pid()} | 'ignore' | {'error', any()})).
 -spec(publish/3 :: (queue_name(), message(), bool()) -> 'ok').
 -spec(fetch/1 :: (queue_name()) ->
-             ('empty' | {message(), non_neg_integer(),
-                         bool(), {msg_id(), seq_id()}, non_neg_integer()})).
+             ('empty' | {{message(), non_neg_integer(),
+                          bool(), {msg_id(), seq_id()}}, non_neg_integer()})).
 -spec(phantom_fetch/1 :: (queue_name()) ->
-             ( 'empty' | {msg_id(), bool(), bool(), {msg_id(), seq_id()},
+             ( 'empty' | {{msg_id(), bool(), bool(), {msg_id(), seq_id()}},
                           non_neg_integer()})).
 -spec(prefetch/1 :: (queue_name()) -> 'ok'). 
 -spec(ack/2 :: (queue_name(), [{msg_id(), seq_id()}]) -> 'ok').
@@ -541,7 +541,7 @@ handle_cast({prefetch, Q, From}, State) ->
 	    true ->
 		case internal_fetch(Q, false, true, true, State1) of
 		    {ok, empty, State2} -> State2;
-		    {ok, {_MsgId, _IsPersistent, _Delivered, _MsgSeqId, _Rem},
+		    {ok, {{_MsgId, _IsPersistent, _Delivered, _MsgSeqId}, _Rem},
 		     State2} -> State2
 		end;
 	    false -> State1
@@ -898,15 +898,7 @@ internal_fetch(Q, ReadMsg, FakeDeliver, Advance,
                                           {Q, ReadSeqId+1, WriteSeqId});
                        false -> true
                    end,
-            {ok,
-             case Result of
-                 {MsgId, IsPersistent, Delivered, {MsgId, ReadSeqId}} ->
-                     {MsgId, IsPersistent, Delivered, {MsgId, ReadSeqId},
-                      Remaining};
-                 {Message, BodySize, Delivered, {MsgId, ReadSeqId}} ->
-                     {Message, BodySize, Delivered, {MsgId, ReadSeqId},
-                      Remaining}
-             end, State1}
+            {ok, {Result, Remaining}, State1}
     end.
 
 internal_foldl(Q, Fun, Init, State) ->
@@ -966,7 +958,7 @@ internal_read_message(Q, ReadSeqId, ReadMsg, FakeDeliver, ForceInCache, State) -
 internal_auto_ack(Q, State) ->
     case internal_fetch(Q, false, false, true, State) of
         {ok, empty, State1} -> {ok, State1};
-        {ok, {_MsgId, _IsPersistent, _Delivered, MsgSeqId, _Remaining},
+        {ok, {{_MsgId, _IsPersistent, _Delivered, MsgSeqId}, _Remaining},
          State1} ->
             remove_messages(Q, [MsgSeqId], true, State1)
     end.        
diff --git a/src/rabbit_mixed_queue.erl b/src/rabbit_mixed_queue.erl
index f865b19a9c..cb34750f06 100644
--- a/src/rabbit_mixed_queue.erl
+++ b/src/rabbit_mixed_queue.erl
@@ -327,7 +327,7 @@ publish_delivered(Msg =
             %% 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} =
+            {{MsgId, IsPersistent, true, AckTag}, 0} =
                 rabbit_disk_queue:phantom_fetch(Q),
             {ok, AckTag, State1};
         false ->
@@ -354,7 +354,7 @@ fetch(State = #mqstate { msg_buf = MsgBuf, queue = Q,
             AckTag =
                 case IsDurable andalso IsPersistent of
                     true ->
-                        {MsgId, IsPersistent, IsDelivered, AckTag1, _PRem}
+                        {{MsgId, IsPersistent, IsDelivered, AckTag1}, _PRem}
                             = rabbit_disk_queue:phantom_fetch(Q),
                         AckTag1;
                     false ->
@@ -372,8 +372,8 @@ fetch(State = #mqstate { msg_buf = MsgBuf, queue = Q,
              State1 #mqstate { msg_buf = MsgBuf1 }};
         _ when Prefetcher == undefined ->
             State2 = dec_queue_length(1, State1),
-            {Msg = #basic_message { is_persistent = IsPersistent },
-             _Size, IsDelivered, AckTag, _PersistRem}
+            {{Msg = #basic_message { is_persistent = IsPersistent },
+              _Size, IsDelivered, AckTag}, _PersistRem}
                 = rabbit_disk_queue:fetch(Q),
             AckTag1 = maybe_ack(Q, IsDurable, IsPersistent, AckTag),
             {{Msg, IsDelivered, AckTag1, Rem}, State2};
diff --git a/src/rabbit_queue_prefetcher.erl b/src/rabbit_queue_prefetcher.erl
index f22aa6afe3..bab1b0c8cb 100644
--- a/src/rabbit_queue_prefetcher.erl
+++ b/src/rabbit_queue_prefetcher.erl
@@ -139,10 +139,10 @@
 %% disk_queue. This msg won't go away and the disk_queue will
 %% eventually find it. However, when it does, it'll simply read the
 %% next message from the queue (which could now be empty), possibly
-%% populate the cache (no harm done), mark the message as deleted (oh
-%% well, not a spec violation, and better than the alternative) and
-%% try and call prefetcher:publish(Msg) which will result in an error,
-%% which the disk_queue catches, as the publish call is to a
+%% populate the cache (no harm done), mark the message as delivered
+%% (oh well, not a spec violation, and better than the alternative)
+%% and try and call prefetcher:publish(Msg) which will result in an
+%% error, which the disk_queue catches, as the publish call is to a
 %% non-existant process. However, the state of the queue has not been
 %% altered so the mixed_queue will be able to fetch this message as if
 %% it had never been prefetched.
@@ -179,8 +179,8 @@
 start_link(Queue, Count) ->
     gen_server2:start_link(?MODULE, [Queue, Count, self()], []).
 
-publish(Prefetcher, Obj = { #basic_message {}, _Size, _IsDelivered,
-                            _AckTag, _Remaining }) ->
+publish(Prefetcher, Obj = {{ #basic_message {}, _Size, _IsDelivered, _AckTag},
+                           _Remaining }) ->
     gen_server2:call(Prefetcher, {publish, Obj}, infinity);
 publish(Prefetcher, empty) ->
     gen_server2:call(Prefetcher, publish_empty, infinity).
@@ -206,8 +206,8 @@ init([Q, Count, QPid]) ->
     {ok, State, infinity, {backoff, ?HIBERNATE_AFTER_MIN,
                            ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
-handle_call({publish, { Msg = #basic_message {},
-                        _Size, IsDelivered, AckTag, _Remaining }},
+handle_call({publish, { { Msg = #basic_message {}, _Size, IsDelivered, AckTag},
+                        _Remaining }},
 	    DiskQueue, State =
 	    #pstate { fetched_count = Fetched, target_count = Target,
 		      msg_buf = MsgBuf, buf_length = Length, queue = Q
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index a9d6080b04..acf3eb7f2f 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -876,7 +876,7 @@ rdq_time_tx_publish_commit_deliver_ack(Qs, MsgCount, MsgSizeBytes) ->
           [[fun() -> [begin SeqIds =
                                 [begin
                                      Remaining = MsgCount - N,
-                                     {Message, _TSize, false, SeqId,
+                                     {{Message, _TSize, false, SeqId},
                                       Remaining} = rabbit_disk_queue:fetch(Q),
                                      ok = rdq_match_message(Message, N, Msg, MsgSizeBytes),
                                      SeqId
@@ -923,7 +923,7 @@ rdq_stress_gc(MsgCount) ->
         lists:foldl(
           fun (MsgId, Acc) ->
                   Remaining = MsgCount - MsgId,
-                  {Message, _TSize, false, SeqId, Remaining} =
+                  {{Message, _TSize, false, SeqId}, Remaining} =
                       rabbit_disk_queue:fetch(q),
                   ok = rdq_match_message(Message, MsgId, Msg, MsgSizeBytes),
                   dict:store(MsgId, SeqId, Acc)
@@ -951,7 +951,7 @@ rdq_test_startup_with_queue_gaps() ->
     %% deliver first half
     Seqs = [begin
                 Remaining = Total - N,
-                {Message, _TSize, false, SeqId, Remaining} =
+                {{Message, _TSize, false, SeqId}, Remaining} =
                     rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
@@ -973,7 +973,7 @@ rdq_test_startup_with_queue_gaps() ->
     %% lists:seq(2,500,2) already delivered
     Seqs2 = [begin
                  Remaining = round(Total - ((Half + N)/2)),
-                 {Message, _TSize, true, SeqId, Remaining} =
+                 {{Message, _TSize, true, SeqId}, Remaining} =
                      rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
@@ -983,7 +983,7 @@ rdq_test_startup_with_queue_gaps() ->
     %% and now fetch the rest
     Seqs3 = [begin
                  Remaining = Total - N,
-                 {Message, _TSize, false, SeqId, Remaining} =
+                 {{Message, _TSize, false, SeqId}, Remaining} =
                      rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
@@ -1008,7 +1008,7 @@ rdq_test_redeliver() ->
     %% deliver first half
     Seqs = [begin
                 Remaining = Total - N,
-                {Message, _TSize, false, SeqId, Remaining} =
+                {{Message, _TSize, false, SeqId}, Remaining} =
                     rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
@@ -1029,7 +1029,7 @@ rdq_test_redeliver() ->
     %% every-other-from-the-first-half
     Seqs2 = [begin
                  Remaining = round(Total - N + (Half/2)),
-                 {Message, _TSize, false, SeqId, Remaining} =
+                 {{Message, _TSize, false, SeqId}, Remaining} =
                      rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
@@ -1037,7 +1037,7 @@ rdq_test_redeliver() ->
     rabbit_disk_queue:tx_commit(q, [], Seqs2),
     Seqs3 = [begin
                  Remaining = round((Half - N) / 2) - 1,
-                 {Message, _TSize, true, SeqId, Remaining} =
+                 {{Message, _TSize, true, SeqId}, Remaining} =
                      rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId
@@ -1061,7 +1061,7 @@ rdq_test_purge() ->
     %% deliver first half
     Seqs = [begin
                 Remaining = Total - N,
-                {Message, _TSize, false, SeqId, Remaining} =
+                {{Message, _TSize, false, SeqId}, Remaining} =
                     rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
@@ -1272,7 +1272,7 @@ rdq_test_disk_queue_modes() ->
     ok = rabbit_disk_queue:tx_commit(q, CommitHalf2, []),
     Seqs = [begin
                 Remaining = Total - N,
-                {Message, _TSize, false, SeqId, Remaining} =
+                {{Message, _TSize, false, SeqId}, Remaining} =
                     rabbit_disk_queue:fetch(q),
                 ok = rdq_match_message(Message, N, Msg, 256),
                 SeqId
@@ -1282,7 +1282,7 @@ rdq_test_disk_queue_modes() ->
     io:format("To RAM Disk done~n", []),
     Seqs2 = [begin
                  Remaining = Total - N,
-                 {Message, _TSize, false, SeqId, Remaining} =
+                 {{Message, _TSize, false, SeqId}, Remaining} =
                      rabbit_disk_queue:fetch(q),
                  ok = rdq_match_message(Message, N, Msg, 256),
                  SeqId