Remove dispatch pause, it's now controller in the polling strategy

ruby-shoryuken · Feb 17, 2017 · 574f927 · 574f927
1 parent 9395b98
commit 574f927
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 13 deletions.
diff --git a/lib/shoryuken/manager.rb b/lib/shoryuken/manager.rb
@@ -2,7 +2,8 @@ module Shoryuken
   class Manager
     include Util
 
-    BATCH_LIMIT = 10
+    BATCH_LIMIT        = 10
+    HEARTBEAT_INTERVAL = 0.1
 
     def initialize(fetcher, polling_strategy)
       @count = Shoryuken.options.fetch(:concurrency, 25)
@@ -17,7 +18,9 @@ def initialize(fetcher, polling_strategy)
       @fetcher = fetcher
       @polling_strategy = polling_strategy
 
-      @heartbeat = Concurrent::TimerTask.new(run_now: true, execution_interval: 0.1, timeout_interval: 60) { dispatch }
+      @heartbeat = Concurrent::TimerTask.new(run_now: true,
+                                             execution_interval: HEARTBEAT_INTERVAL,
+                                             timeout_interval: 60) { dispatch }
 
       @pool = Concurrent::FixedThreadPool.new(@count, max_queue: @count)
     end
@@ -59,15 +62,10 @@ def dispatch
       return if @done.true?
       return unless @dispatching.make_true
 
-      logger.debug { "Ready: #{ready}, Busy: #{busy}, Active Queues: #{@polling_strategy.active_queues}" }
-
-      if ready == 0
-        return logger.debug { 'Pausing fetcher, because all processors are busy' }
-      end
+      return if ready == 0
+      return unless (queue = @polling_strategy.next_queue)
 
-      unless (queue = @polling_strategy.next_queue)
-        return logger.debug { 'Pausing fetcher, because all queues are paused' }
-      end
+      logger.debug { "Ready: #{ready}, Busy: #{busy}, Active Queues: #{@polling_strategy.active_queues}" }
 
       batched_queue?(queue) ? dispatch_batch(queue) : dispatch_single_messages(queue)
     ensure

diff --git a/lib/shoryuken/polling.rb b/lib/shoryuken/polling.rb
@@ -65,7 +65,6 @@ def delay
     end
 
     class WeightedRoundRobin < BaseStrategy
-
       def initialize(queues)
         @initial_queues = queues
         @queues = queues.dup.uniq
@@ -129,7 +128,6 @@ def queue_weight(queues, queue)
     end
 
     class StrictPriority < BaseStrategy
-
       def initialize(queues)
         # Priority ordering of the queues, highest priority first
         @queues = queues
@@ -178,7 +176,7 @@ def next_active_queue
           return queue unless queue_paused?(queue)
         end
 
-        return nil
+        nil
       end
 
       def queues_unpaused_since?