refactor possibly_unblock

1 files changed, 19 insertions, 18 deletions

diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index fb60a043..0530d650 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -644,25 +644,26 @@ remove_consumers(ChPid, Queue, QName) ->
 
 possibly_unblock(State, ChPid, Update) ->
     case lookup_ch(ChPid) of
-        not_found ->
+        not_found -> State;
+        C         -> possibly_unblock(State, Update(C))
+    end.
+
+possibly_unblock(State, C = #cr{limiter = Limiter}) ->
+    IsChBlocked = is_ch_blocked(C),
+    case lists:partition(
+           fun({_ChPid, #consumer{tag = CTag}}) ->
+                   IsChBlocked orelse
+                       rabbit_limiter:is_consumer_blocked(Limiter, CTag)
+           end, queue:to_list(C#cr.blocked_consumers)) of
+        {_, []} ->
+            update_ch_record(C),
             State;
-        C ->
-            C1 = #cr{limiter = Limiter} = Update(C),
-            {Blocked, Unblocked} =
-                lists:partition(
-                  fun({_ChPid, #consumer{tag = CTag}}) ->
-                         is_ch_blocked(C1) orelse
-                              rabbit_limiter:is_consumer_blocked(Limiter, CTag)
-                  end, queue:to_list(C1#cr.blocked_consumers)),
-            case Unblocked of
-                [] -> update_ch_record(C1),
-                      State;
-                _  -> update_ch_record(
-                        C1#cr{blocked_consumers = queue:from_list(Blocked)}),
-                      AC1 = queue:join(State#q.active_consumers,
-                                       queue:from_list(Unblocked)),
-                      run_message_queue(State#q{active_consumers = AC1})
-            end
+        {Blocked, Unblocked} ->
+            BlockedQ   = queue:from_list(Blocked),
+            UnblockedQ = queue:from_list(Unblocked),
+            update_ch_record(C#cr{blocked_consumers = BlockedQ}),
+            AC1 = queue:join(State#q.active_consumers, UnblockedQ),
+            run_message_queue(State#q{active_consumers = AC1})
     end.
 
 should_auto_delete(#q{q = #amqqueue{auto_delete = false}}) -> false;