Merge pull request #13374 from rabbitmq/rabbitmq-server-12743

By @noxdafox: Revival of #9620 Pass the message to rabbit_backing_queue:discard callback

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -648,7 +648,7 @@ discard(#delivery{confirm = Confirm,
                true  -> confirm_messages([MsgId], MTC, QName);
                false -> MTC
            end,
-    BQS1 = BQ:discard(MsgId, SenderPid, BQS),
+    BQS1 = BQ:discard(Msg, SenderPid, BQS),
     {BQS1, MTC1}.
 
 run_message_queue(ActiveConsumersChanged, State) ->
@@ -828,7 +828,7 @@ send_reject_publish(#delivery{confirm = true,
                                              amqqueue:get_name(Q), MsgSeqNo),
 
     MTC1 = maps:remove(MsgId, MTC),
-    BQS1 = BQ:discard(MsgId, SenderPid, BQS),
+    BQS1 = BQ:discard(Msg, SenderPid, BQS),
     State#q{ backing_queue_state = BQS1, msg_id_to_channel = MTC1 };
 send_reject_publish(#delivery{confirm = false}, State) ->
     State.

deps/rabbit/src/rabbit_backing_queue.erl

@@ -105,7 +105,7 @@
 
 %% Called to inform the BQ about messages which have reached the
 %% queue, but are not going to be further passed to BQ.
--callback discard(rabbit_types:msg_id(), pid(), state()) -> state().
+-callback discard(rabbit_types:basic_message(), pid(), state()) -> state().
 
 %% Return ids of messages which have been confirmed since the last
 %% invocation of this function (or initialisation).

deps/rabbit/src/rabbit_priority_queue.erl

@@ -220,22 +220,12 @@ publish_delivered(Msg, MsgProps, ChPid,
                   State = #passthrough{bq = BQ, bqs = BQS}) ->
     ?passthrough2(publish_delivered(Msg, MsgProps, ChPid, BQS)).
 
-%% TODO this is a hack. The BQ api does not give us enough information
-%% here - if we had the Msg we could look at its priority and forward
-%% to the appropriate sub-BQ. But we don't so we are stuck.
-%%
-%% But fortunately VQ ignores discard/4, so we can too, *assuming we
-%% are talking to VQ*. discard/4 is used by HA, but that's "above" us
-%% (if in use) so we don't break that either, just some hypothetical
-%% alternate BQ implementation.
-discard(_MsgId, _ChPid, State = #state{}) ->
-    State;
-    %% We should have something a bit like this here:
-    %% pick1(fun (_P, BQSN) ->
-    %%               BQ:discard(MsgId, ChPid, BQSN)
-    %%       end, Msg, State);
-discard(MsgId, ChPid, State = #passthrough{bq = BQ, bqs = BQS}) ->
-    ?passthrough1(discard(MsgId, ChPid, BQS)).
+discard(Msg, ChPid, State = #state{bq = BQ}) ->
+    pick1(fun (_P, BQSN) ->
+                  BQ:discard(Msg, ChPid, BQSN)
+          end, Msg, State);
+discard(Msg, ChPid, State = #passthrough{bq = BQ, bqs = BQS}) ->
+    ?passthrough1(discard(Msg, ChPid, BQS)).
 
 drain_confirmed(State = #state{bq = BQ}) ->
     fold_append2(fun (_P, BQSN) -> BQ:drain_confirmed(BQSN) end, State);

deps/rabbit/src/rabbit_variable_queue.erl

@@ -544,7 +544,7 @@ publish_delivered(Msg, MsgProps, ChPid, State) ->
                            State),
     {SeqId, a(maybe_update_rates(State1))}.
 
-discard(_MsgId, _ChPid, State) -> State.
+discard(_Msg, _ChPid, State) -> State.
 
 drain_confirmed(State = #vqstate { confirmed = C }) ->
     case sets:is_empty(C) of