add pause and resume to ThreadedConsumerPool

Author: Aaron Patterson and Terence Lee

lib/resque/threaded_consumer_pool.rb

@@ -1,19 +1,17 @@
 module Resque
   class ThreadedConsumerPool
-
     def initialize(queue, size)
       @queue = queue
       @size  = size
       @threads = []
       @consumers = []
     end
 
-
     def start
       stop
       @consumers.clear
       @threads = @size.times.map {
-        c = Consumer.new(@queue)
+        c = build_consumer @queue
         @consumers << c
         Thread.new { c.consume }
       }
@@ -34,5 +32,19 @@ def term
     def kill
       @threads.each { |t| t.kill }
     end
+
+    def pause
+      @consumers.each { |c| c.pause }
+    end
+
+    def resume
+      @consumers.each { |c| c.resume }
+    end
+
+    private
+
+    def build_consumer(queue)
+      Consumer.new queue
+    end
   end
 end

test/threaded_consumer_pool_test.rb

@@ -64,10 +64,6 @@ def run
       TermJob.clear
     end
 
-    after do
-      @tp.join
-    end
-
     it "processes work" do
       Resque.consumer_timeout = 1
       5.times { @write << Actionable.new }
@@ -87,6 +83,7 @@ def run
       sleep 1
       @tp.stop
       assert @write.empty?
+      @tp.join
     end
 
     it "terms the consumers" do
@@ -119,5 +116,28 @@ def run
       assert TermJob.termed.empty?
       assert @read.empty?
     end
+
+    it "pauses and resumes" do
+      paused = []
+      resumed = []
+
+      @tp = Class.new(ThreadedConsumerPool) {
+        define_method(:build_consumer) { |q|
+          super(q).extend(Module.new {
+            define_method(:pause) { paused << self; super() }
+            define_method(:resume) { resumed << self; super() }
+          })
+        }
+      }.new(@read, 1)
+
+      @tp.start
+      @tp.pause
+      assert_equal 1, paused.length
+
+      @tp.resume
+      assert_equal 1, resumed.length
+      @tp.stop
+      @tp.join
+    end
   end
 end