added publish_delivered/2 which deals with adding a message, when the queue is empty which we already know has been sent out to a consumer, so it's really just a case of writing to disk the message, and index pub and deliver entries iff the message is persistent

1 files changed, 17 insertions, 1 deletions

diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 0ffc2adcc3..e7c546e545 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -31,7 +31,7 @@
 
 -module(rabbit_variable_queue).
 
--export([init/1, publish/2, set_queue_ram_duration_target/2,
+-export([init/1, publish/2, publish_delivered/2, 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_commit/3, do_tx_commit/3]).
@@ -143,6 +143,22 @@ init(QueueName) ->
 publish(Msg, State) ->
     publish(Msg, false, false, State).
 
+publish_delivered(Msg = #basic_message { guid = MsgId,
+                                         is_persistent = IsPersistent },
+                  State = #vqstate { len = 0, index_state = IndexState,
+                                     next_seq_id = SeqId }) ->
+    case maybe_write_msg_to_disk(false, false, Msg) of
+        true ->
+            {true, IndexState1} =
+                maybe_write_index_to_disk(false, IsPersistent, MsgId, SeqId,
+                                          true, IndexState),
+            {{ack_index_and_store, MsgId, SeqId},
+             State #vqstate { index_state = IndexState1,
+                              next_seq_id = SeqId + 1 }};
+        false ->
+            {ack_not_on_disk, State}
+    end.
+
 set_queue_ram_duration_target(
   DurationTarget, State = #vqstate { avg_egress_rate = EgressRate,
                                      target_ram_msg_count = TargetRamMsgCount