Move bulk_requeue back to class method.

lib/sidekiq/fetch.rb

@@ -47,10 +47,6 @@ module Sidekiq
       end
     end
 
-    def bulk_requeue(in_progress)
-      @strategy.bulk_requeue(in_progress.values)
-    end
-
     def handle_fetch_exception(ex)
       if !@down
         logger.error("Error fetching message: #{ex}")
@@ -94,7 +90,9 @@ module Sidekiq
       UnitOfWork.new(*work) if work
     end
 
-    def bulk_requeue(inprogress)
+    # By leaving this as a class method, it can be pluggable and used by the Manager actor. Making it
+    # an instance method will make it async to the Fetcher actor
+    def self.bulk_requeue(inprogress, options)
       Sidekiq.logger.debug { "Re-queueing terminated jobs" }
       jobs_to_requeue = {}
       inprogress.each do |unit_of_work|

lib/sidekiq/manager.rb

@@ -23,6 +23,7 @@ module Sidekiq
 
     def initialize(options={})
       logger.debug { options.inspect }
+      @options = options
       @count = options[:concurrency] || 25
       @done_callback = nil
 
@@ -207,7 +208,7 @@ module Sidekiq
       # contract says that jobs are run AT LEAST once. Process termination
       # is delayed until we're certain the jobs are back in Redis because
       # it is worse to lose a job than to run it twice.
-      @fetcher.bulk_requeue(@in_progress.values)
+      Sidekiq::Fetcher.strategy.bulk_requeue(@in_progress.values, @options)
       @in_progress.clear
     end
   end

test/test_fetch.rb

@@ -36,7 +36,7 @@ class TestFetcher < Sidekiq::Test
       assert_equal 0, q1.size
       assert_equal 0, q2.size
       uow = Sidekiq::BasicFetch::UnitOfWork
-      Sidekiq::BasicFetch.new({:queues => []}).bulk_requeue([uow.new('fuzzy:queue:foo', 'bob'), uow.new('fuzzy:queue:foo', 'bar'), uow.new('fuzzy:queue:bar', 'widget')])
+      Sidekiq::BasicFetch.bulk_requeue([uow.new('fuzzy:queue:foo', 'bob'), uow.new('fuzzy:queue:foo', 'bar'), uow.new('fuzzy:queue:bar', 'widget')], {:queues => []})
       assert_equal 2, q1.size
       assert_equal 1, q2.size
     end