Maybe done. Maybe time to start hacking out mq and dq. Maybe.

2 files changed, 35 insertions, 22 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index d55a38d79c..3228655257 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -39,7 +39,7 @@
 -export([list/1, info/1, info/2, info_all/1, info_all/2]).
 -export([claim_queue/2]).
 -export([basic_get/3, basic_consume/8, basic_cancel/4]).
--export([notify_sent/2, unblock/2]).
+-export([notify_sent/2, unblock/2, tx_commit_callback/3]).
 -export([commit_all/2, rollback_all/2, notify_down_all/2, limit_all/3]).
 -export([on_node_down/1]).
 -export([set_storage_mode/2]).
@@ -63,6 +63,8 @@
 -type(qfun(A) :: fun ((amqqueue()) -> A)).
 -type(ok_or_errors() ::
       'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}).
+-type(seq_id() :: non_neg_integer()).
+-type(acktag() :: ('ack_not_on_disk' | {'ack_index_and_store', msg_id(), seq_id()})).
 
 -spec(start/0 :: () -> 'ok').
 -spec(recover/0 :: () -> {'ok', [amqqueue()]}).
@@ -104,6 +106,7 @@
 -spec(basic_cancel/4 :: (amqqueue(), pid(), ctag(), any()) -> 'ok').
 -spec(notify_sent/2 :: (pid(), pid()) -> 'ok').
 -spec(unblock/2 :: (pid(), pid()) -> 'ok').
+-spec(tx_commit_callback/3 :: (pid(), [message()], [acktag()]) -> 'ok').
 -spec(set_storage_mode/2 :: (pid(), ('oppressed' | 'liberated')) -> 'ok').
 -spec(internal_declare/2 :: (amqqueue(), boolean()) -> amqqueue()).
 -spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()).
@@ -320,6 +323,9 @@ notify_sent(QPid, ChPid) ->
 unblock(QPid, ChPid) ->
     gen_server2:pcast(QPid, 8, {unblock, ChPid}).
 
+tx_commit_callback(QPid, Pubs, AckTags) ->
+    gen_server2:pcast(QPid, 8, {tx_commit_callback, Pubs, AckTags}).
+
 internal_delete(QueueName) ->
     rabbit_misc:execute_mnesia_transaction(
       fun () ->
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 97e8141fe6..2b02466979 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -34,7 +34,7 @@
 -export([init/1, publish/3, set_queue_ram_duration_target/2,
          remeasure_egress_rate/1, fetch/1, ack/2, len/1, is_empty/1,
          maybe_start_prefetcher/1, purge/1, delete/1, requeue/2,
-         tx_publish/2, tx_rollback/2]).
+         tx_publish/2, tx_rollback/2, tx_commit/3, do_tx_commit/3]).
 
 %%----------------------------------------------------------------------------
 
@@ -145,9 +145,9 @@ publish(Msg, IsDelivered, State) ->
 
 publish(Msg, IsDelivered, PersistentMsgsAlreadyOnDisk,
         State = #vqstate { next_seq_id = SeqId, len = Len }) ->
-    publish(test_keep_msg_in_ram(SeqId, State), Msg, SeqId, IsDelivered,
-            PersistentMsgsAlreadyOnDisk,
-            State #vqstate { next_seq_id = SeqId + 1, len = Len + 1 }).
+    {SeqId, publish(test_keep_msg_in_ram(SeqId, State), Msg, SeqId, IsDelivered,
+                    PersistentMsgsAlreadyOnDisk,
+                    State #vqstate { next_seq_id = SeqId + 1, len = Len + 1 })}.
 
 set_queue_ram_duration_target(
   DurationTarget, State = #vqstate { avg_egress_rate = EgressRate,
@@ -312,21 +312,28 @@ tx_rollback(Pubs, State) ->
     ok = rabbit_msg_store:remove(persistent_msg_ids(Pubs)),
     State.
 
-%% tx_commit(Pubs, AckTags, State) ->
-%%     case persistent_msg_ids(Pubs) of
-%%         [] ->
-%%             do_tx_commit(Pubs, AckTags, State);
-%%         PersistentMsgIds ->
-%%             ok = rabbit_msg_store:sync(
-%%                    PersistentMsgIds,
-%%                    fun () -> ok = rabbit_amqqueue:tx_commit_callback(
-%%                                     self(), Pubs, AckTags)
-%%                    end),
-%%             State
-%%     end.
-
-%% do_tx_commit(Pubs, AckTags, State) ->
-%%     lists:foldl(fun (Msg, StateN) -> publish(Msg, false, StateN) end, State, Pubs).
+tx_commit(Pubs, AckTags, State) ->
+    case persistent_msg_ids(Pubs) of
+        [] ->
+            do_tx_commit(Pubs, AckTags, State);
+        PersistentMsgIds ->
+            ok = rabbit_msg_store:sync(
+                   PersistentMsgIds,
+                   fun () -> ok = rabbit_amqqueue:tx_commit_callback(
+                                    self(), Pubs, AckTags)
+                   end),
+            State
+    end.
+
+do_tx_commit(Pubs, AckTags, State) ->
+    {_PubSeqIds, State1} =
+        lists:foldl(
+          fun (Msg, {SeqIdsAcc, StateN}) ->
+                  {SeqId, StateN1} = publish(Msg, false, true, StateN),
+                  {[SeqId | SeqIdsAcc], StateN1}
+          end, {[], State}, Pubs),
+    %% TODO need to do something here about syncing the queue index, PubSeqIds
+    ack(AckTags, State1).
 
 %%----------------------------------------------------------------------------
 
@@ -695,9 +702,9 @@ maybe_push_alphas_to_betas(Generator, Consumer, Q, State =
           #alpha { msg = Msg = #basic_message { guid = MsgId,
                                                 is_persistent = IsPersistent },
                    seq_id = SeqId, is_delivered = IsDelivered,
-                   msg_on_disk = MsgOnDisk, index_on_disk = IndexOnDisk }},
+                   index_on_disk = IndexOnDisk }},
          Qa} ->
-            true = maybe_write_msg_to_disk(true, MsgOnDisk, Msg),
+            true = maybe_write_msg_to_disk(true, true, Msg),
             Beta = #beta { msg_id = MsgId, seq_id = SeqId,
                            is_persistent = IsPersistent,
                            is_delivered = IsDelivered,