diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit_fifo.erl | 10 | ||||
| -rw-r--r-- | src/rabbit_quorum_queue.erl | 4 |
2 files changed, 7 insertions, 7 deletions
diff --git a/src/rabbit_fifo.erl b/src/rabbit_fifo.erl index 247085460e..ef98dc5f98 100644 --- a/src/rabbit_fifo.erl +++ b/src/rabbit_fifo.erl @@ -332,7 +332,7 @@ apply(Meta, #discard{msg_ids = MsgIds, consumer_id = ConsumerId}, case Cons0 of #{ConsumerId := Con0} -> Discarded = maps:with(MsgIds, Con0#consumer.checked_out), - Effects = dead_letter_effects(Discarded, State0, []), + Effects = dead_letter_effects(rejected, Discarded, State0, []), complete_and_checkout(Meta, MsgIds, ConsumerId, Con0, Effects, State0); _ -> @@ -948,7 +948,7 @@ drop_head(#state{ra_indexes = Indexes0} = State0, Effects0) -> Indexes = rabbit_fifo_index:delete(RaftIdxToDrop, Indexes0), Bytes = message_size(Msg), State = add_bytes_drop(Bytes, State1#state{ra_indexes = Indexes}), - Effects = dead_letter_effects(maps:put(none, FullMsg, #{}), + Effects = dead_letter_effects(maxlen, maps:put(none, FullMsg, #{}), State, Effects0), {State, Effects}; {{'$prefix_msg', Bytes}, State1} -> @@ -1099,15 +1099,15 @@ complete_and_checkout(#{index := IncomingRaftIdx} = Meta, MsgIds, ConsumerId, % settle metrics are incremented separately update_smallest_raft_index(IncomingRaftIdx, Indexes0, State, Effects). -dead_letter_effects(_Discarded, +dead_letter_effects(_Reason, _Discarded, #state{dead_letter_handler = undefined}, Effects) -> Effects; -dead_letter_effects(Discarded, +dead_letter_effects(Reason, Discarded, #state{dead_letter_handler = {Mod, Fun, Args}}, Effects) -> DeadLetters = maps:fold(fun(_, {_, {_, {_Header, Msg}}}, % MsgId, MsgIdID, RaftId, Header - Acc) -> [{rejected, Msg} | Acc] + Acc) -> [{Reason, Msg} | Acc] end, [], Discarded), [{mod_call, Mod, Fun, Args ++ [DeadLetters]} | Effects]. diff --git a/src/rabbit_quorum_queue.erl b/src/rabbit_quorum_queue.erl index 725d8086d9..f83ad11b50 100644 --- a/src/rabbit_quorum_queue.erl +++ b/src/rabbit_quorum_queue.erl @@ -756,8 +756,8 @@ i(pid, Q) when ?is_amqqueue(Q) -> {Name, _} = amqqueue:get_pid(Q), whereis(Name); i(messages, Q) when ?is_amqqueue(Q) -> - {Name, _} = amqqueue:get_pid(Q), - quorum_messages(Name); + QName = amqqueue:get_name(Q), + quorum_messages(QName); i(messages_ready, Q) when ?is_amqqueue(Q) -> QName = amqqueue:get_name(Q), case ets:lookup(queue_coarse_metrics, QName) of |