Use new handle_info rabbit_backing_queue callback

Author: lukebakken

src/rabbit_amqqueue_process.erl

@@ -1455,8 +1455,9 @@ handle_info({bump_credit, Msg}, State = #q{backing_queue       = BQ,
     credit_flow:handle_bump_msg(Msg),
     noreply(State#q{backing_queue_state = BQ:resume(BQS)});
 handle_info(bump_reduce_memory_use, State = #q{backing_queue       = BQ,
-                                               backing_queue_state = BQS}) ->
-    noreply(State#q{backing_queue_state = BQ:resume(BQS)});
+                                               backing_queue_state = BQS0}) ->
+    BQS1 = BQ:handle_info(bump_reduce_memory_use, BQS0),
+    noreply(State#q{backing_queue_state = BQ:resume(BQS1)});
 
 handle_info(Info, State) ->
     {stop, {unhandled_info, Info}, State}.

src/rabbit_mirror_queue_master.erl

@@ -24,7 +24,7 @@
          dropwhile/2, fetchwhile/4, set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1, resume/1,
          msg_rates/1, info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4]).
+         zip_msgs_and_acks/4, handle_info/2]).
 
 -export([start/2, stop/1, delete_crashed/1]).
 
@@ -447,6 +447,10 @@ handle_pre_hibernate(State = #state { backing_queue       = BQ,
                                       backing_queue_state = BQS }) ->
     State #state { backing_queue_state = BQ:handle_pre_hibernate(BQS) }.
 
+handle_info(Msg, State = #state { backing_queue       = BQ,
+                                  backing_queue_state = BQS }) ->
+    State #state { backing_queue_state = BQ:handle_info(Msg, BQS) }.
+
 resume(State = #state { backing_queue       = BQ,
                         backing_queue_state = BQS }) ->
     State #state { backing_queue_state = BQ:resume(BQS) }.

src/rabbit_priority_queue.erl

@@ -41,7 +41,7 @@
          set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1,
          handle_pre_hibernate/1, resume/1, msg_rates/1,
          info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4]).
+         zip_msgs_and_acks/4, handle_info/2]).
 
 -record(state, {bq, bqss, max_priority}).
 -record(passthrough, {bq, bqs}).
@@ -393,6 +393,9 @@ handle_pre_hibernate(State = #state{bq = BQ}) ->
 handle_pre_hibernate(State = #passthrough{bq = BQ, bqs = BQS}) ->
     ?passthrough1(handle_pre_hibernate(BQS)).
 
+handle_info(Msg, State = #passthrough{bq = BQ, bqs = BQS}) ->
+    ?passthrough1(handle_info(Msg, BQS)).
+
 resume(State = #state{bq = BQ}) ->
     foreach1(fun (_P, BQSN) -> BQ:resume(BQSN) end, State);
 resume(State = #passthrough{bq = BQ, bqs = BQS}) ->

src/rabbit_variable_queue.erl

@@ -26,7 +26,7 @@
          set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1,
          handle_pre_hibernate/1, resume/1, msg_rates/1,
          info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4,  multiple_routing_keys/0]).
+         zip_msgs_and_acks/4,  multiple_routing_keys/0, handle_info/2]).
 
 -export([start/2, stop/1]).
 
@@ -912,10 +912,10 @@ timeout(State = #vqstate { index_state = IndexState }) ->
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
 
-resume(State = #vqstate{waiting_bump = true}) ->
-    resume(State#vqstate{waiting_bump = false});
-resume(State) ->
-    a(reduce_memory_use(State)).
+handle_info(bump_reduce_memory_use, State = #vqstate{ waiting_bump = true }) ->
+    State#vqstate{ waiting_bump = false }.
+
+resume(State) -> a(reduce_memory_use(State)).
 
 msg_rates(#vqstate { rates = #rates { in  = AvgIngressRate,
                                       out = AvgEgressRate } }) ->
@@ -2505,11 +2505,11 @@ reduce_memory_use(State = #vqstate {
     garbage_collect(),
     State3.
 
-maybe_bump_reduce_memory_use(State = #vqstate{waiting_bump = true}) ->
+maybe_bump_reduce_memory_use(State = #vqstate{ waiting_bump = true }) ->
     State;
 maybe_bump_reduce_memory_use(State) ->
     self() ! bump_reduce_memory_use,
-    State#vqstate{waiting_bump = true}.
+    State#vqstate{ waiting_bump = true }.
 
 limit_ram_acks(0, State) ->
     {0, ui(State)};

test/channel_operation_timeout_test_queue.erl

@@ -26,7 +26,7 @@
          set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1,
          handle_pre_hibernate/1, resume/1, msg_rates/1,
          info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         start/2, stop/1, zip_msgs_and_acks/4]).
+         start/2, stop/1, zip_msgs_and_acks/4, handle_info/2]).
 
 %%----------------------------------------------------------------------------
 %% This test backing queue follows the variable queue implementation, with
@@ -285,6 +285,9 @@ timeout(State) ->
 handle_pre_hibernate(State) ->
     rabbit_variable_queue:handle_pre_hibernate(State).
 
+handle_info(Msg, State) ->
+    rabbit_variable_queue:handle_info(Msg, State).
+
 resume(State) -> rabbit_variable_queue:resume(State).
 
 msg_rates(State) ->