issue #22: fix publish_delivered callback with deduplication queue

When the queue is empty, the message is passed straight to the first
consumer to save one round trip time.

In such cases, if message acknowledgement is enabled,
the `publish_delivered` callback is called instead of `fetch`.

If de-duplication is enabled, we need to ensure a `dqack`
is returned instead of a simple acknowledgement tag.

Signed-off-by: Matteo Cafasso <[email protected]>

Author: noxdafox

lib/rabbit_message_deduplication_queue.ex

@@ -190,12 +190,21 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
     passthrough1(state, do: batch_publish(batch, pid, flow, qs))
   end
 
+  # Optimization for cases in which the queue is empty and the message
+  # is delivered straight to the client. Acknowledgement is enabled.
   def publish_delivered(message, message_properties, pid, flow, state) do
-    dqstate(queue_state: qs) = state
+    dqstate(queue: queue, queue_state: qs) = state
 
-    passthrough2(state) do
+    {ack_tag, state} = passthrough2(state) do
       publish_delivered(message, message_properties, pid, flow, qs)
     end
+
+    if duplicate?(queue) do
+      head = Common.message_header(message, "x-deduplication-header")
+      {dqack(tag: ack_tag, header: head), state}
+    else
+      {ack_tag, state}
+    end
   end
 
   def batch_publish_delivered(batch, pid, flow, state) do