Merge pull request #1765 from rabbitmq/remove-dead-letter-process

Remove dead letter process

Author: kjnilsson

src/rabbit_amqqueue_process.erl

@@ -954,13 +954,11 @@ dead_letter_maxlen_msg(X, State = #q{backing_queue = BQ}) ->
 
 dead_letter_msgs(Fun, Reason, X, State = #q{dlx_routing_key     = RK,
                                             backing_queue_state = BQS,
-                                            backing_queue       = BQ,
-                                            q = #amqqueue{ name = Resource } }) ->
-    #resource{virtual_host = VHost} = Resource,
+                                            backing_queue       = BQ}) ->
     QName = qname(State),
     {Res, Acks1, BQS1} =
         Fun(fun (Msg, AckTag, Acks) ->
-                    rabbit_vhost_dead_letter:publish(VHost, X, RK, QName, [{Reason, Msg}]),
+                    rabbit_dead_letter:publish(Msg, Reason, X, RK, QName),
                     [AckTag | Acks]
             end, [], BQS),
     {_Guids, BQS2} = BQ:ack(Acks1, BQS1),

src/rabbit_dead_letter.erl

@@ -16,7 +16,7 @@
 
 -module(rabbit_dead_letter).
 
--export([publish/6]).
+-export([publish/5]).
 
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
@@ -26,20 +26,17 @@
 -type reason() :: 'expired' | 'rejected' | 'maxlen'.
 
 -spec publish(rabbit_types:message(), reason(), rabbit_types:exchange(),
-              'undefined' | binary(), rabbit_amqqueue:name(),
-              #{Name :: atom() => rabbit_fifo_client:state()}) -> 'ok'.
+              'undefined' | binary(), rabbit_amqqueue:name()) -> 'ok'.
 
 %%----------------------------------------------------------------------------
 
-publish(Msg, Reason, X, RK, QName, QueueStates0) ->
+publish(Msg, Reason, X, RK, QName) ->
     DLMsg = make_msg(Msg, Reason, X#exchange.name, RK, QName),
     Delivery = rabbit_basic:delivery(false, false, DLMsg, undefined),
     {Queues, Cycles} = detect_cycles(Reason, DLMsg,
                                      rabbit_exchange:route(X, Delivery)),
     lists:foreach(fun log_cycle_once/1, Cycles),
-    {_, _, QueueStates} = rabbit_amqqueue:deliver(rabbit_amqqueue:lookup(Queues),
-                                                  Delivery, QueueStates0),
-    QueueStates.
+    rabbit_amqqueue:deliver(rabbit_amqqueue:lookup(Queues), Delivery).
 
 make_msg(Msg = #basic_message{content       = Content,
                               exchange_name = Exchange,

src/rabbit_quorum_queue.erl

@@ -23,7 +23,7 @@
 -export([credit/4]).
 -export([purge/1]).
 -export([stateless_deliver/2, deliver/3]).
--export([dead_letter_publish/5]).
+-export([dead_letter_publish/4]).
 -export([queue_name/1]).
 -export([cluster_state/1, status/2]).
 -export([cancel_consumer_handler/3, cancel_consumer/3]).
@@ -523,11 +523,10 @@ delete_member(#amqqueue{pid = {RaName, _}, name = QName}, Node) ->
     end.
 
 %%----------------------------------------------------------------------------
-dlx_mfa(#amqqueue{name = Resource} = Q) ->
-    #resource{virtual_host = VHost} = Resource,
+dlx_mfa(Q) ->
     DLX = init_dlx(args_policy_lookup(<<"dead-letter-exchange">>, fun res_arg/2, Q), Q),
     DLXRKey = args_policy_lookup(<<"dead-letter-routing-key">>, fun res_arg/2, Q),
-    {?MODULE, dead_letter_publish, [VHost, DLX, DLXRKey, Q#amqqueue.name]}.
+    {?MODULE, dead_letter_publish, [DLX, DLXRKey, Q#amqqueue.name]}.
 
 init_dlx(undefined, _Q) ->
     undefined;
@@ -545,10 +544,12 @@ args_policy_lookup(Name, Resolve, Q = #amqqueue{arguments = Args}) ->
         {PolVal,    {_Type, ArgVal}} -> Resolve(PolVal, ArgVal)
     end.
 
-dead_letter_publish(_, undefined, _, _, _) ->
+dead_letter_publish(undefined, _, _, _) ->
     ok;
-dead_letter_publish(VHost, X, RK, QName, ReasonMsgs) ->
-    rabbit_vhost_dead_letter:publish(VHost, X, RK, QName, ReasonMsgs).
+dead_letter_publish(X, RK, QName, ReasonMsgs) ->
+    {ok, Exchange} = rabbit_exchange:lookup(X),
+    [rabbit_dead_letter:publish(Msg, Reason, Exchange, RK, QName)
+     || {Reason, Msg} <- ReasonMsgs].
 
 %% TODO escape hack
 qname_to_rname(#resource{virtual_host = <<"/">>, name = Name}) ->

src/rabbit_variable_queue.erl

@@ -488,12 +488,10 @@ start(VHost, DurableQueues) ->
                           Ref = proplists:get_value(persistent_ref, Terms),
                           Ref =/= undefined
                       end],
-    start_dead_letter_process(VHost),
     start_msg_store(VHost, ClientRefs, StartFunState),
     {ok, AllTerms}.
 
 stop(VHost) ->
-    ok = rabbit_vhost_dead_letter:stop(VHost),
     ok = stop_msg_store(VHost),
     ok = rabbit_queue_index:stop(VHost).
 
@@ -517,14 +515,6 @@ do_start_msg_store(VHost, Type, Refs, StartFunState) ->
             exit({error, Error})
     end.
 
-start_dead_letter_process(VHost) ->
-    case rabbit_vhost_dead_letter:start(VHost) of
-        {ok, _} ->
-            rabbit_log:info("Started dead letter process for vhost '~s'~n", [VHost]);
-        Err ->
-            exit(Err)
-    end.
-
 abbreviated_type(?TRANSIENT_MSG_STORE)  -> transient;
 abbreviated_type(?PERSISTENT_MSG_STORE) -> persistent.