merge default into bug20337

9 files changed, 317 insertions, 78 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index b3e92b6918..041f0b4a23 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -316,34 +316,47 @@ assert_args_equivalence(#amqqueue{name = QueueName, arguments = Args},
       Args, RequiredArgs, QueueName,
       [<<"x-expires">>, <<"x-message-ttl">>, <<"x-ha-policy">>]).
 
-check_declare_arguments(QueueName, Args) ->
-    [case Fun(rabbit_misc:table_lookup(Args, Key), Args) of
+check_declare_arguments(QueueName = #resource{virtual_host = VHostPath},
+                        Args) ->
+    [case Fun(rabbit_misc:table_lookup(Args, Key), Args, VHostPath) of
          ok             -> ok;
          {error, Error} -> rabbit_misc:protocol_error(
                              precondition_failed,
                              "invalid arg '~s' for ~s: ~255p",
                              [Key, rabbit_misc:rs(QueueName), Error])
-     end || {Key, Fun} <-
-                [{<<"x-expires">>,     fun check_integer_argument/2},
-                 {<<"x-message-ttl">>, fun check_integer_argument/2},
-                 {<<"x-ha-policy">>,   fun check_ha_policy_argument/2}]],
+     end ||
+        {Key, Fun} <-
+                [{<<"x-expires">>,     fun check_integer_argument/3},
+                 {<<"x-message-ttl">>, fun check_integer_argument/3},
+                 {<<"x-ha-policy">>,   fun check_ha_policy_argument/3},
+                 {<<"x-dead-letter-exchange">>, fun check_exchange_argument/3}]],
     ok.
 
-check_integer_argument(undefined, _Args) ->
+check_integer_argument(undefined, _Args, _VHostPath) ->
     ok;
-check_integer_argument({Type, Val}, _Args) when Val > 0 ->
+check_integer_argument({Type, Val}, _Args, _VHostPath) when Val > 0 ->
     case lists:member(Type, ?INTEGER_ARG_TYPES) of
         true  -> ok;
         false -> {error, {unacceptable_type, Type}}
     end;
-check_integer_argument({_Type, Val}, _Args) ->
+check_integer_argument({_Type, Val}, _Args, _VHostPath) ->
     {error, {value_zero_or_less, Val}}.
 
-check_ha_policy_argument(undefined, _Args) ->
+check_exchange_argument(undefined, _Args, _VHostPath) ->
     ok;
-check_ha_policy_argument({longstr, <<"all">>}, _Args) ->
+check_exchange_argument({longstr, Val}, _Args, VHostPath) ->
+    try rabbit_misc:r(VHostPath, exchange, Val)
+    of _Exchange -> ok
+    catch _:_ -> {error, {invalid_exchange_name, Val}}
+    end;
+check_exchange_argument({Type, _Val}, _Args, _VHostPath) ->
+    {error, {unacceptable_type, Type}}.
+
+check_ha_policy_argument(undefined, _Args, _VHostPath) ->
+    ok;
+check_ha_policy_argument({longstr, <<"all">>}, _Args, _VHostPath) ->
     ok;
-check_ha_policy_argument({longstr, <<"nodes">>}, Args) ->
+check_ha_policy_argument({longstr, <<"nodes">>}, Args, _VHostPath) ->
     case rabbit_misc:table_lookup(Args, <<"x-ha-policy-params">>) of
         undefined ->
             {error, {require, 'x-ha-policy-params'}};
@@ -359,9 +372,9 @@ check_ha_policy_argument({longstr, <<"nodes">>}, Args) ->
         {Type, _} ->
             {error, {ha_nodes_policy_params_not_array_of_longstr, Type}}
     end;
-check_ha_policy_argument({longstr, Policy}, _Args) ->
+check_ha_policy_argument({longstr, Policy}, _Args, _VHostPath) ->
     {error, {invalid_ha_policy, Policy}};
-check_ha_policy_argument({Type, _}, _Args) ->
+check_ha_policy_argument({Type, _}, _Args, _VHostPath) ->
     {error, {unacceptable_type, Type}}.
 
 list() ->
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index ba20b35524..a60dc19914 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -49,7 +49,12 @@
             stats_timer,
             msg_id_to_channel,
             ttl,
-            ttl_timer_ref
+            ttl_timer_ref,
+            publish_seqno,
+            unconfirmed,
+            blocked_op,
+            queue_monitors,
+            dlx
            }).
 
 -record(consumer, {tag, ack_required}).
@@ -129,6 +134,11 @@ init(Q) ->
                rate_timer_ref      = undefined,
                expiry_timer_ref    = undefined,
                ttl                 = undefined,
+               dlx                 = undefined,
+               publish_seqno       = 1,
+               unconfirmed         = gb_trees:empty(),
+               blocked_op          = undefined,
+               queue_monitors      = dict:new(),
                msg_id_to_channel   = gb_trees:empty()},
     {ok, rabbit_event:init_stats_timer(State, #q.stats_timer), hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
@@ -151,6 +161,10 @@ init_with_backing_queue_state(Q = #amqqueue{exclusive_owner = Owner}, BQ, BQS,
                rate_timer_ref      = RateTRef,
                expiry_timer_ref    = undefined,
                ttl                 = undefined,
+               publish_seqno       = 1,
+               unconfirmed         = gb_trees:empty(),
+               blocked_op          = undefined,
+               queue_monitors      = dict:new(),
                msg_id_to_channel   = MTC},
     State1 = requeue_and_run(AckTags, process_args(
                                         rabbit_event:init_stats_timer(
@@ -218,12 +232,17 @@ process_args(State = #q{q = #amqqueue{arguments = Arguments}}) ->
                             undefined    -> State1
                         end
                 end, State, [{<<"x-expires">>,     fun init_expires/2},
-                             {<<"x-message-ttl">>, fun init_ttl/2}]).
+                             {<<"x-message-ttl">>, fun init_ttl/2},
+                             {<<"x-dead-letter-exchange">>, fun init_dlx/2}]).
 
 init_expires(Expires, State) -> ensure_expiry_timer(State#q{expires = Expires}).
 
 init_ttl(TTL, State) -> drop_expired_messages(State#q{ttl = TTL}).
 
+init_dlx(DLX, State = #q{q = #amqqueue{name = #resource{
+                                         virtual_host = VHostPath}}}) ->
+    State#q{dlx = rabbit_misc:r(VHostPath, exchange, DLX)}.
+
 terminate_shutdown(Fun, State) ->
     State1 = #q{backing_queue_state = BQS} =
         stop_sync_timer(stop_rate_timer(State)),
@@ -462,7 +481,7 @@ confirm_messages(MsgIds, State = #q{msg_id_to_channel = MTC}) ->
                           {CMs, MTC0}
                   end
           end, {gb_trees:empty(), MTC}, MsgIds),
-    rabbit_misc:gb_trees_foreach(fun rabbit_channel:confirm/2, CMs),
+    rabbit_misc:gb_trees_foreach(fun rabbit_misc:confirm_to_sender/2, CMs),
     State#q{msg_id_to_channel = MTC1}.
 
 should_confirm_message(#delivery{msg_seq_no = undefined}, _State) ->
@@ -501,7 +520,7 @@ attempt_delivery(Delivery = #delivery{sender     = ChPid,
                  State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
     Confirm = should_confirm_message(Delivery, State),
     case Confirm of
-        immediately -> rabbit_channel:confirm(ChPid, [MsgSeqNo]);
+        immediately -> rabbit_misc:confirm_to_sender(ChPid, [MsgSeqNo]);
         _           -> ok
     end,
     case BQ:is_duplicate(Message, BQS) of
@@ -677,10 +696,11 @@ calculate_msg_expiry(TTL)       -> now_micros() + (TTL * 1000).
 drop_expired_messages(State = #q{ttl = undefined}) ->
     State;
 drop_expired_messages(State = #q{backing_queue_state = BQS,
-                                 backing_queue = BQ}) ->
+                                 backing_queue       = BQ}) ->
     Now = now_micros(),
     BQS1 = BQ:dropwhile(
              fun (#message_properties{expiry = Expiry}) -> Now > Expiry end,
+             mk_dead_letter_fun(expired, State),
              BQS),
     ensure_ttl_timer(State#q{backing_queue_state = BQS1}).
 
@@ -697,6 +717,149 @@ ensure_ttl_timer(State = #q{backing_queue       = BQ,
 ensure_ttl_timer(State) ->
     State.
 
+mk_dead_letter_fun(_Reason, #q{dlx = undefined}) ->
+    fun(_MsgLookupFun, _AckTag, BQS) -> BQS end;
+mk_dead_letter_fun(Reason, _State) ->
+    fun(MsgLookupFun, AckTag, BQS) ->
+            {Msg, BQS1} = MsgLookupFun(BQS),
+            gen_server2:cast(self(), {dead_letter, {Msg, AckTag}, Reason}),
+            BQS1
+    end.
+
+dead_letter_deleted_queue(undefined, State = #q{dlx = undefined}) ->
+    {stop, normal, State};
+dead_letter_deleted_queue(_From, State = #q{dlx                 = undefined,
+                                            backing_queue_state = BQS,
+                                            backing_queue       = BQ}) ->
+    {stop, normal, {ok, BQ:len(BQS)}, State};
+dead_letter_deleted_queue(From, State = #q{backing_queue_state = BQS,
+                                           backing_queue       = BQ}) ->
+    case BQ:len(BQS) of
+        0 -> dead_letter_deleted_queue(From, State#q{dlx = undefined});
+        _ -> BQS1 = BQ:dropwhile(fun (_) -> true end,
+                                 mk_dead_letter_fun(queue_deleted, State),
+                                 BQS),
+             noreply(State#q{blocked_op = {delete, {From, BQ:len(BQS)}},
+                             backing_queue_state = BQS1})
+    end.
+
+dead_letter_msg(Msg, AckTag, Reason,
+                State = #q{publish_seqno       = MsgSeqNo,
+                           unconfirmed         = UC,
+                           dlx                 = DLX,
+                           backing_queue       = BQ,
+                           backing_queue_state = BQS}) ->
+    rabbit_exchange:lookup_or_die(DLX),
+
+    {ok, _, QPids} =
+        rabbit_basic:publish(
+          rabbit_basic:delivery(
+            false, false, make_dead_letter_msg(DLX, Reason, Msg, State),
+            MsgSeqNo)),
+    State1 = lists:foldl(fun monitor_queue/2, State, QPids),
+    State2 = State1#q{publish_seqno = MsgSeqNo + 1},
+    case QPids of
+        [] -> {_, BQS1} = BQ:ack([AckTag], undefined, BQS),
+              cleanup_after_confirm(State2#q{backing_queue_state = BQS1});
+        _  -> noreply(State2#q{
+                        unconfirmed = gb_trees:insert(
+                                        MsgSeqNo, {gb_sets:from_list(QPids),
+                                                   AckTag}, UC)})
+    end.
+
+monitor_queue(QPid, State = #q{queue_monitors = QMons}) ->
+    case dict:is_key(QPid, QMons) of
+        true  -> State;
+        false -> State#q{queue_monitors =
+                             dict:store(QPid, erlang:monitor(process, QPid),
+                                        QMons)}
+    end.
+
+handle_queue_down(QPid, State = #q{queue_monitors = QMons,
+                                   unconfirmed    = UC}) ->
+    case dict:find(QPid, QMons) of
+        error ->
+            noreply(State);
+        {ok, _} ->
+            #resource{name = QName} = qname(State),
+            rabbit_log:info("DLQ ~p (for ~p) died~n", [QPid, QName]),
+            MsgSeqNos = [MsgSeqNo ||
+                            {MsgSeqNo, {QPids, _}} <- gb_trees:to_list(UC),
+                            gb_sets:is_member(QPid, QPids)],
+            handle_confirm(MsgSeqNos, QPid,
+                           State#q{queue_monitors = dict:erase(QPid, QMons)})
+    end.
+
+handle_confirm(MsgSeqNos, QPid, State = #q{unconfirmed         = UC,
+                                           backing_queue       = BQ,
+                                           backing_queue_state = BQS}) ->
+    {BQS3, UC3} =
+        lists:foldl(
+          fun (MsgSeqNo, {BQS1, UC1}) ->
+                  {QPids, AckTag} = gb_trees:get(MsgSeqNo, UC1),
+                  QPids1 = gb_sets:delete(QPid, QPids),
+                  case gb_sets:is_empty(QPids1) of
+                      true  -> {_Guids, BQS2} =
+                                   BQ:ack([AckTag], undefined, BQS1),
+                               {BQS2, gb_trees:delete(MsgSeqNo, UC1)};
+                      false -> {BQS1, gb_trees:update(MsgSeqNo,
+                                                      {QPids1, AckTag}, UC1)}
+                  end
+          end, {BQS, UC}, MsgSeqNos),
+    cleanup_after_confirm(State#q{unconfirmed         = UC3,
+                                  backing_queue_state = BQS3}).
+
+cleanup_after_confirm(State = #q{blocked_op  = Op,
+                                 unconfirmed = UC}) ->
+    State1 = State#q{blocked_op = undefined},
+    case {gb_trees:is_empty(UC), Op} of
+        {true, {purge, {From, Count}}} ->
+            gen_server2:reply(From, {ok, Count}),
+            noreply(State1);
+        {true, {delete, {From, Count}}} ->
+            case From of
+                undefined -> ok;
+                _         -> gen_server2:reply(From, {ok, Count})
+            end,
+            {stop, normal, State1};
+        _ ->
+            noreply(State1#q{blocked_op = Op})
+    end.
+
+make_dead_letter_msg(DLX, Reason, Msg = #basic_message{content = Content},
+                     State) ->
+
+    Content1 = #content{
+      properties = Props = #'P_basic'{headers = Headers}} =
+        rabbit_binary_parser:ensure_content_decoded(Content),
+
+    #resource{name = QName} = qname(State),
+
+    DeathTable = {table, [{<<"x-death-reason">>, longstr,
+                           list_to_binary(atom_to_list(Reason))},
+                          {<<"x-death-queue">>, longstr, QName}]},
+
+    Headers1 =
+        case Headers of
+            undefined ->
+                [{<<"x-death">>, array, [DeathTable]}];
+            _ ->
+                case rabbit_misc:table_lookup(Headers, <<"x-death">>) of
+                    undefined ->
+                        [{<<"x-death">>, array, [DeathTable]} | Headers];
+                    {array, Prior} ->
+                        rabbit_misc:set_table_value(
+                          Headers, <<"x-death">>, array, [DeathTable | Prior])
+                end
+        end,
+    Content2 =
+        rabbit_binary_generator:clear_encoded_content(
+          Content1#content{properties = Props#'P_basic'{headers = Headers1}}),
+
+    Msg#basic_message{exchange_name = DLX, id = rabbit_guid:guid(),
+                      content = Content2}.
+
+
 now_micros() -> timer:now_diff(now(), {0,0,0}).
 
 infos(Items, State) ->
@@ -994,7 +1157,7 @@ handle_call(stat, _From, State) ->
         drop_expired_messages(ensure_expiry_timer(State)),
     reply({ok, BQ:len(BQS), active_consumer_count()}, State1);
 
-handle_call({delete, IfUnused, IfEmpty}, _From,
+handle_call({delete, IfUnused, IfEmpty}, From,
             State = #q{backing_queue_state = BQS, backing_queue = BQ}) ->
     IsEmpty = BQ:is_empty(BQS),
     IsUnused = is_unused(State),
@@ -1004,14 +1167,24 @@ handle_call({delete, IfUnused, IfEmpty}, _From,
         IfUnused and not(IsUnused) ->
             reply({error, in_use}, State);
         true ->
-            {stop, normal, {ok, BQ:len(BQS)}, State}
+            dead_letter_deleted_queue(From, State)
     end;
 
-handle_call(purge, _From, State = #q{backing_queue = BQ,
-                                     backing_queue_state = BQS}) ->
+handle_call(purge, _From, State = #q{backing_queue       = BQ,
+                                     backing_queue_state = BQS,
+                                     dlx                 = undefined}) ->
     {Count, BQS1} = BQ:purge(BQS),
     reply({ok, Count}, State#q{backing_queue_state = BQS1});
 
+handle_call(purge, From, State = #q{backing_queue       = BQ,
+                                    backing_queue_state = BQS}) ->
+    BQS1 = BQ:dropwhile(
+             fun (_) -> true end,
+             mk_dead_letter_fun(queue_purged, State),
+             BQS),
+    noreply(State#q{backing_queue_state = BQS1,
+                    blocked_op          = {purge, {From, BQ:len(BQS)}}});
+
 handle_call({requeue, AckTags, ChPid}, From, State) ->
     gen_server2:reply(From, ok),
     noreply(subtract_acks(
@@ -1030,7 +1203,8 @@ handle_cast({ack, AckTags, ChPid}, State) ->
               ChPid, AckTags, State,
               fun (State1 = #q{backing_queue       = BQ,
                                backing_queue_state = BQS}) ->
-                      {_Guids, BQS1} = BQ:ack(AckTags, BQS),
+                      {_Guids, BQS1} =
+                          BQ:ack(AckTags, undefined, BQS),
                       State1#q{backing_queue_state = BQS1}
               end));
 
@@ -1041,13 +1215,15 @@ handle_cast({reject, AckTags, Requeue, ChPid}, State) ->
                                backing_queue_state = BQS}) ->
                       case Requeue of
                           true  -> requeue_and_run(AckTags, State1);
-                          false -> {_Guids, BQS1} = BQ:ack(AckTags, BQS),
+                          false -> Fun = mk_dead_letter_fun(rejected, State),
+                                   {_Guids, BQS1} =
+                                       BQ:ack(AckTags, Fun, BQS),
                                    State1#q{backing_queue_state = BQS1}
                       end
               end));
 
 handle_cast(delete_immediately, State) ->
-    {stop, normal, State};
+    dead_letter_deleted_queue(undefined, State);
 
 handle_cast({unblock, ChPid}, State) ->
     noreply(
@@ -1098,12 +1274,18 @@ handle_cast(force_event_refresh, State = #q{exclusive_consumer = Exclusive}) ->
         {Ch, CTag} -> [{Ch, CTag, AckRequired}] = consumers(State),
                       emit_consumer_created(Ch, CTag, true, AckRequired)
     end,
-    noreply(State).
+    noreply(State);
+
+handle_cast({confirm, MsgSeqNos, QPid}, State) ->
+    handle_confirm(MsgSeqNos, QPid, State);
+
+handle_cast({dead_letter, {Msg, AckTag}, Reason}, State) ->
+    dead_letter_msg(Msg, AckTag, Reason, State).
 
 handle_info(maybe_expire, State) ->
     case is_unused(State) of
         true  -> ?LOGDEBUG("Queue lease expired for ~p~n", [State#q.q]),
-                 {stop, normal, State};
+                 dead_letter_deleted_queue(undefined, State);
         false -> noreply(ensure_expiry_timer(State))
     end;
 
@@ -1128,8 +1310,8 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason},
     {stop, normal, State};
 handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) ->
     case handle_ch_down(DownPid, State) of
-        {ok, NewState}   -> noreply(NewState);
-        {stop, NewState} -> {stop, normal, NewState}
+        {ok, State1}   -> handle_queue_down(DownPid, State1);
+        {stop, State1} -> {stop, normal, State1}
     end;
 
 handle_info(update_ram_duration, State = #q{backing_queue = BQ,
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index c3b322ee6d..72c00e3d01 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -95,15 +95,19 @@ behaviour_info(callbacks) ->
      {drain_confirmed, 1},
 
      %% Drop messages from the head of the queue while the supplied
-     %% predicate returns true.
-     {dropwhile, 2},
+     %% predicate returns true. A callback function is supplied
+     %% allowing callers access to messages that are about to be
+     %% dropped.
+     {dropwhile, 3},
 
      %% Produce the next message.
      {fetch, 2},
 
      %% Acktags supplied are for messages which can now be forgotten
-     %% about. Must return 1 msg_id per Ack, in the same order as Acks.
-     {ack, 2},
+     %% about. Must return 1 msg_id per Ack, in the same order as
+     %% Acks. A callback function is supplied allowing callers to
+     %% access messages that are being acked.
+     {ack, 3},
 
      %% Reinsert messages into the queue which have already been
      %% delivered and were pending acknowledgement.
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 9b2fe28ce8..9df862c635 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -21,7 +21,7 @@
 -behaviour(gen_server2).
 
 -export([start_link/10, do/2, do/3, flush/1, shutdown/1]).
--export([send_command/2, deliver/4, flushed/2, confirm/2]).
+-export([send_command/2, deliver/4, flushed/2]).
 -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]).
 -export([refresh_config_local/0, ready_for_close/1]).
 -export([force_event_refresh/0]).
@@ -85,7 +85,6 @@
         (pid(), rabbit_types:ctag(), boolean(), rabbit_amqqueue:qmsg())
         -> 'ok').
 -spec(flushed/2 :: (pid(), pid()) -> 'ok').
--spec(confirm/2 ::(pid(), [non_neg_integer()]) -> 'ok').
 -spec(list/0 :: () -> [pid()]).
 -spec(list_local/0 :: () -> [pid()]).
 -spec(info_keys/0 :: () -> rabbit_types:info_keys()).
@@ -128,9 +127,6 @@ deliver(Pid, ConsumerTag, AckRequired, Msg) ->
 flushed(Pid, QPid) ->
     gen_server2:cast(Pid, {flushed, QPid}).
 
-confirm(Pid, MsgSeqNos) ->
-    gen_server2:cast(Pid, {confirm, MsgSeqNos, self()}).
-
 list() ->
     rabbit_misc:append_rpc_all_nodes(rabbit_mnesia:running_clustered_nodes(),
                                      rabbit_channel, list_local, []).
diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl
index f60562efe6..4b54d82171 100644
--- a/src/rabbit_mirror_queue_master.erl
+++ b/src/rabbit_mirror_queue_master.erl
@@ -17,8 +17,8 @@
 -module(rabbit_mirror_queue_master).
 
 -export([init/3, terminate/2, delete_and_terminate/2,
-         purge/1, publish/4, publish_delivered/5, fetch/2, ack/2,
-         requeue/2, len/1, is_empty/1, drain_confirmed/1, dropwhile/2,
+         purge/1, publish/4, publish_delivered/5, fetch/2, ack/3,
+         requeue/2, len/1, is_empty/1, drain_confirmed/1, dropwhile/3,
          set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1,
          status/1, invoke/3, is_duplicate/2, discard/3]).
@@ -172,12 +172,13 @@ publish_delivered(AckRequired, Msg = #basic_message { id = MsgId }, MsgProps,
      ensure_monitoring(ChPid, State #state { backing_queue_state = BQS1,
                                              ack_msg_id          = AM1 })}.
 
-dropwhile(Fun, State = #state { gm                  = GM,
-                                backing_queue       = BQ,
-                                backing_queue_state = BQS,
-                                set_delivered       = SetDelivered }) ->
+dropwhile(Pred, MsgFun,
+          State = #state{gm                  = GM,
+                         backing_queue       = BQ,
+                         set_delivered       = SetDelivered,
+                         backing_queue_state = BQS }) ->
     Len = BQ:len(BQS),
-    BQS1 = BQ:dropwhile(Fun, BQS),
+    BQS1 = BQ:dropwhile(Pred, MsgFun, BQS),
     Dropped = Len - BQ:len(BQS1),
     SetDelivered1 = lists:max([0, SetDelivered - Dropped]),
     ok = gm:broadcast(GM, {set_length, BQ:len(BQS1)}),
@@ -235,15 +236,15 @@ fetch(AckRequired, State = #state { gm                  = GM,
                              ack_msg_id    = AM1 }}
     end.
 
-ack(AckTags, State = #state { gm                  = GM,
-                              backing_queue       = BQ,
-                              backing_queue_state = BQS,
-                              ack_msg_id          = AM }) ->
-    {MsgIds, BQS1} = BQ:ack(AckTags, BQS),
+ack(AckTags, MsgFun, State = #state { gm                  = GM,
+                                      backing_queue       = BQ,
+                                      backing_queue_state = BQS,
+                                      ack_msg_id          = AM }) ->
+    {MsgIds, BQS1} = BQ:ack(AckTags, MsgFun, BQS),
     AM1 = lists:foldl(fun dict:erase/2, AM, AckTags),
     case MsgIds of
         [] -> ok;
-        _  -> ok = gm:broadcast(GM, {ack, MsgIds})
+        _  -> ok = gm:broadcast(GM, {ack, MsgFun, MsgIds})
     end,
     {MsgIds, State #state { backing_queue_state = BQS1,
                             ack_msg_id          = AM1 }}.
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index 7182042d44..a8c2006d14 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -417,7 +417,7 @@ confirm_messages(MsgIds, State = #state { msg_id_status = MS }) ->
                           Acc
                   end
           end, {gb_trees:empty(), MS}, MsgIds),
-    rabbit_misc:gb_trees_foreach(fun rabbit_channel:confirm/2, CMs),
+    rabbit_misc:gb_trees_foreach(fun rabbit_misc:confirm_to_sender/2, CMs),
     State #state { msg_id_status = MS1 }.
 
 handle_process_result({ok,   State}) -> noreply(State);
@@ -649,7 +649,7 @@ maybe_enqueue_message(
         {ok, {confirmed, ChPid}} ->
             %% BQ has confirmed it but we didn't know what the
             %% msg_seq_no was at the time. We do now!
-            ok = rabbit_channel:confirm(ChPid, [MsgSeqNo]),
+            ok = rabbit_misc:confirm_to_sender(ChPid, [MsgSeqNo]),
             SQ1 = remove_from_pending_ch(MsgId, ChPid, SQ),
             State1 #state { sender_queues = SQ1,
                             msg_id_status = dict:erase(MsgId, MS) };
@@ -666,7 +666,7 @@ maybe_enqueue_message(
                       msg_id_status =
                           dict:store(MsgId, {published, ChPid, MsgSeqNo}, MS) };
                 immediately ->
-                    ok = rabbit_channel:confirm(ChPid, [MsgSeqNo]),
+                    ok = rabbit_misc:confirm_to_sender(ChPid, [MsgSeqNo]),
                     SQ1 = remove_from_pending_ch(MsgId, ChPid, SQ),
                     State1 #state { msg_id_status = dict:erase(MsgId, MS),
                                     sender_queues = SQ1 }
@@ -728,7 +728,7 @@ process_instruction(
                         {MQ2, sets:add_element(MsgId, PendingCh),
                          dict:store(MsgId, {published, ChPid, MsgSeqNo}, MS)};
                     immediately ->
-                        ok = rabbit_channel:confirm(ChPid, [MsgSeqNo]),
+                        ok = rabbit_misc:confirm_to_sender(ChPid, [MsgSeqNo]),
                         {MQ2, PendingCh, MS}
                 end;
             {{value, {#delivery {}, _EnqueueOnPromotion}}, _MQ2} ->
@@ -818,12 +818,12 @@ process_instruction({fetch, AckRequired, MsgId, Remaining},
                  %% we must be shorter than the master
                  State
          end};
-process_instruction({ack, MsgIds},
+process_instruction({ack, MsgFun, MsgIds},
                     State = #state { backing_queue       = BQ,
                                      backing_queue_state = BQS,
                                      msg_id_ack          = MA }) ->
     {AckTags, MA1} = msg_ids_to_acktags(MsgIds, MA),
-    {MsgIds1, BQS1} = BQ:ack(AckTags, BQS),
+    {MsgIds1, BQS1} = BQ:ack(AckTags, MsgFun, BQS),
     [] = MsgIds1 -- MsgIds, %% ASSERTION
     {ok, State #state { msg_id_ack          = MA1,
                         backing_queue_state = BQS1 }};
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 88192e8f5b..d1006c0e0a 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -28,6 +28,7 @@
 -export([enable_cover/0, report_cover/0]).
 -export([enable_cover/1, report_cover/1]).
 -export([start_cover/1]).
+-export([confirm_to_sender/2]).
 -export([throw_on_error/2, with_exit_handler/2, filter_exit_map/2]).
 -export([with_user/2, with_user_and_vhost/3]).
 -export([execute_mnesia_transaction/1]).
@@ -371,6 +372,9 @@ report_coverage_percentage(File, Cov, NotCov, Mod) ->
                end,
                Mod]).
 
+confirm_to_sender(Pid, MsgSeqNos) ->
+    gen_server2:cast(Pid, {confirm, MsgSeqNos, self()}).
+
 throw_on_error(E, Thunk) ->
     case Thunk() of
         {error, Reason} -> throw({E, Reason});
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 00d46f5a56..2f03b2fb0b 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -2334,7 +2334,9 @@ test_dropwhile(VQ0) ->
     VQ2 = rabbit_variable_queue:dropwhile(
             fun(#message_properties { expiry = Expiry }) ->
                     Expiry =< 5
-            end, VQ1),
+            end,
+            dummy_msg_fun(),
+            VQ1),
 
     %% fetch five now
     VQ3 = lists:foldl(fun (_N, VQN) ->
@@ -2348,13 +2350,17 @@ test_dropwhile(VQ0) ->
 
     VQ4.
 
+dummy_msg_fun() -> fun(_Fun, _Extra, State) -> State end.
+
 test_dropwhile_varying_ram_duration(VQ0) ->
     VQ1 = variable_queue_publish(false, 1, VQ0),
     VQ2 = rabbit_variable_queue:set_ram_duration_target(0, VQ1),
-    VQ3 = rabbit_variable_queue:dropwhile(fun(_) -> false end, VQ2),
+    VQ3 = rabbit_variable_queue:dropwhile(
+            fun(_) -> false end, dummy_msg_fun(), VQ2),
     VQ4 = rabbit_variable_queue:set_ram_duration_target(infinity, VQ3),
     VQ5 = variable_queue_publish(false, 1, VQ4),
-    rabbit_variable_queue:dropwhile(fun(_) -> false end, VQ5).
+    rabbit_variable_queue:dropwhile(
+      fun(_) -> false end, dummy_msg_fun(), VQ5).
 
 test_variable_queue_dynamic_duration_change(VQ0) ->
     SegmentSize = rabbit_queue_index:next_segment_boundary(0),
@@ -2379,7 +2385,7 @@ test_variable_queue_dynamic_duration_change(VQ0) ->
 
     %% drain
     {VQ8, AckTags} = variable_queue_fetch(Len, false, false, Len, VQ7),
-    {_Guids, VQ9} = rabbit_variable_queue:ack(AckTags, VQ8),
+    {_Guids, VQ9} = rabbit_variable_queue:ack(AckTags, undefined, VQ8),
     {empty, VQ10} = rabbit_variable_queue:fetch(true, VQ9),
 
     VQ10.
@@ -2389,7 +2395,7 @@ publish_fetch_and_ack(0, _Len, VQ0) ->
 publish_fetch_and_ack(N, Len, VQ0) ->
     VQ1 = variable_queue_publish(false, 1, VQ0),
     {{_Msg, false, AckTag, Len}, VQ2} = rabbit_variable_queue:fetch(true, VQ1),
-    {_Guids, VQ3} = rabbit_variable_queue:ack([AckTag], VQ2),
+    {_Guids, VQ3} = rabbit_variable_queue:ack([AckTag], undefined, VQ2),
     publish_fetch_and_ack(N-1, Len, VQ3).
 
 test_variable_queue_partial_segments_delta_thing(VQ0) ->
@@ -2423,7 +2429,8 @@ test_variable_queue_partial_segments_delta_thing(VQ0) ->
              {len, HalfSegment + 1}]),
     {VQ8, AckTags1} = variable_queue_fetch(HalfSegment + 1, true, false,
                                            HalfSegment + 1, VQ7),
-    {_Guids, VQ9} = rabbit_variable_queue:ack(AckTags ++ AckTags1, VQ8),
+    {_Guids, VQ9} = rabbit_variable_queue:ack(AckTags ++ AckTags1,
+                                              undefined, VQ8),
     %% should be empty now
     {empty, VQ10} = rabbit_variable_queue:fetch(true, VQ9),
     VQ10.
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 63a0927f77..92c650272f 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -18,7 +18,7 @@
 
 -export([init/3, terminate/2, delete_and_terminate/2,
          purge/1, publish/4, publish_delivered/5, drain_confirmed/1,
-         dropwhile/2, fetch/2, ack/2, requeue/2, len/1, is_empty/1,
+         dropwhile/3, fetch/2, ack/3, requeue/2, len/1, is_empty/1,
          set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1,
          status/1, invoke/3, is_duplicate/2, discard/3,
@@ -581,15 +581,22 @@ drain_confirmed(State = #vqstate { confirmed = C }) ->
                                         confirmed = gb_sets:new() }}
     end.
 
-dropwhile(Pred, State) ->
+dropwhile(Pred, MsgFun, State) ->
     case queue_out(State) of
         {empty, State1} ->
             a(State1);
         {{value, MsgStatus = #msg_status { msg_props = MsgProps }}, State1} ->
-            case Pred(MsgProps) of
-                true ->  {_, State2} = internal_fetch(false, MsgStatus, State1),
-                         dropwhile(Pred, State2);
-                false -> a(in_r(MsgStatus, State1))
+            case {Pred(MsgProps), MsgFun} of
+                {true, undefined} ->
+                    {_, State2} = internal_fetch(false, MsgStatus, State1),
+                    dropwhile(Pred, MsgFun, State2);
+                {true, _} ->
+                    {{_, _, AckTag, _}, State2} =
+                        internal_fetch(true, MsgStatus, State1),
+                    State3 = MsgFun(read_msg_callback(MsgStatus), AckTag, State2),
+                    dropwhile(Pred, MsgFun, State3);
+                {false, _} ->
+                    a(in_r(MsgStatus, State1))
             end
     end.
 
@@ -605,9 +612,27 @@ fetch(AckRequired, State) ->
             {Res, a(State3)}
     end.
 
-ack([], State) ->
+read_msg_callback(#msg_status { msg           = undefined,
+                                msg_id        = MsgId,
+                                is_persistent = IsPersistent }) ->
+    fun(State) -> read_msg_callback1(MsgId, IsPersistent, State) end;
+
+read_msg_callback(#msg_status{ msg = Msg }) ->
+    fun(State) -> {Msg, State} end;
+
+read_msg_callback({IsPersistent, MsgId, _MsgProps}) ->
+    fun(State) -> read_msg_callback1(MsgId, IsPersistent, State) end.
+
+read_msg_callback1(MsgId, IsPersistent,
+                   State = #vqstate{ msg_store_clients = MSCState }) ->
+    {{ok, Msg = #basic_message{}}, MSCState1} =
+        msg_store_read(MSCState, IsPersistent, MsgId),
+    {Msg, State #vqstate { msg_store_clients = MSCState1 }}.
+
+ack([], _Fun, State) ->
     {[], State};
-ack(AckTags, State) ->
+
+ack(AckTags, undefined, State) ->
     {{IndexOnDiskSeqIds, MsgIdsByStore, AllMsgIds},
      State1 = #vqstate { index_state       = IndexState,
                          msg_store_clients = MSCState,
@@ -616,7 +641,7 @@ ack(AckTags, State) ->
         lists:foldl(
           fun (SeqId, {Acc, State2}) ->
                   {MsgStatus, State3} = remove_pending_ack(SeqId, State2),
-                  {accumulate_ack(MsgStatus, Acc), State3}
+                  {accumulate_ack(MsgStatus, Acc),  State3}
           end, {accumulate_ack_init(), State}, AckTags),
     IndexState1 = rabbit_queue_index:ack(IndexOnDiskSeqIds, IndexState),
     [ok = msg_store_remove(MSCState, IsPersistent, MsgIds)
@@ -626,13 +651,20 @@ ack(AckTags, State) ->
     {lists:reverse(AllMsgIds),
      a(State1 #vqstate { index_state      = IndexState1,
                          persistent_count = PCount1,
-                         ack_out_counter  = AckOutCount + length(AckTags) })}.
+                         ack_out_counter  = AckOutCount + length(AckTags) })};
+
+ack(AckTags, MsgFun, State = #vqstate{pending_ack = PA}) ->
+    [begin
+         AckEntry = gb_trees:get(SeqId, PA),
+         MsgFun(read_msg_callback(AckEntry), SeqId, State)
+     end || SeqId <- AckTags],
+    {[], State}.
 
 requeue(AckTags, #vqstate { delta = Delta,
-                                    q3         = Q3,
-                                    q4         = Q4,
-                                    in_counter = InCounter,
-                                    len        = Len } = State) ->
+                            q3         = Q3,
+                            q4         = Q4,
+                            in_counter = InCounter,
+                            len        = Len } = State) ->
     {SeqIds,  Q4a, MsgIds,  State1} = queue_merge(lists:sort(AckTags), Q4, [],
                                                   beta_limit(Q3),
                                                   fun publish_alpha/2, State),