Merge remote-tracking branch 'tenderlove/consumers' into consumers

Author: hone

lib/resque/consumer.rb

@@ -1,17 +1,23 @@
 module Resque
   class Consumer
     class Latch # :nodoc:
-      def initialize
-        @mutex = Mutex.new
-        @cond  = ConditionVariable.new
+      def initialize(count = 1)
+        @count = count
+        @lock  = Monitor.new
+        @cv    = @lock.new_cond
       end
 
       def release
-        @mutex.synchronize { @cond.broadcast }
+        @lock.synchronize do
+          @count -= 1 if @count > 0
+          @cv.broadcast if @count.zero?
+        end
       end
 
       def await
-        @mutex.synchronize { @cond.wait @mutex }
+        @lock.synchronize do
+          @cv.wait_while { @count > 0 }
+        end
       end
     end
 
@@ -57,7 +63,7 @@ def resume
       @paused       = false
       @latch.release
     end
-    
+
     def shutdown
       @should_shutdown = true
     end

test/consumer_test.rb

@@ -13,7 +13,20 @@ def run
         self.class.ran << self
       end
     end
-    
+
+    class Resumer
+      LATCHES = {}
+
+      def initialize(latch)
+        @latch_id          = latch.object_id
+        LATCHES[@latch_id] = latch
+      end
+
+      def run
+        LATCHES[@latch_id].release
+      end
+    end
+
     before do
       Actionable.ran.clear
     end
@@ -28,7 +41,7 @@ def run
       assert_raises Timeout::Error do
         Timeout.timeout(1) { c.consume }
       end
-      
+
       assert_equal 1, Actionable.ran.length
       assert q.empty?
     end
@@ -59,17 +72,19 @@ def run
       q.pop until q.empty?
 
       c = Consumer.new(q, 1)
+      consumed = Consumer::Latch.new
 
       c.pause
       t = Thread.new { c.consume }
-      # wait until queue blocks
-      sleep 1
+      Thread.pass until c.paused?
 
-      q << Actionable.new
+      # A job that unblocks the main thread
+      q << Resumer.new(consumed)
       c.resume
-      # wait until queue blocks
-      Thread.pass until t.status != "sleep"
-      sleep 1
+
+      # wait until consumed
+      consumed.await
+
       assert_equal 0, q.length, 'all jobs should be consumed'
       t.kill
     end