block when popping jobs in Worker

pairing with @tenderlove

Author: hone

lib/resque/worker.rb

@@ -128,7 +128,7 @@ def work(interval = 5.0, &block)
       loop do
         break if shutdown?
 
-        if not paused? and job = reserve
+        if not paused? and job = reserve(interval)
           log "got: #{job.inspect}"
           job.worker = self
           run_hook :before_fork, job
@@ -148,9 +148,8 @@ def work(interval = 5.0, &block)
           @child = nil
         else
           break if interval.zero?
-          log! "Sleeping for #{interval} seconds"
+          log! "Timed out after #{interval} seconds"
           procline paused? ? "Paused" : "Waiting for #{@queues.join(',')}"
-          sleep interval
         end
       end
 
@@ -192,12 +191,18 @@ def perform(job)
 
     # Attempts to grab a job off one of the provided queues. Returns
     # nil if no job can be found.
-    def reserve
+    def reserve(interval = 5.0)
+      interval = interval.to_i
       multi_queue = MultiQueue.new(
         queues.map {|queue| Queue.new(queue, Resque.redis, Resque.coder) },
         Resque.redis)
 
-      queue, job = multi_queue.pop(true)
+      if interval < 1
+        queue, job = multi_queue.pop(true)
+      else
+        queue, job = multi_queue.poll(interval.to_i)
+      end
+
       log! "Found job on #{queue}"
       return Job.new(queue.name, job)
     rescue ThreadError

test/worker_test.rb

@@ -264,6 +264,14 @@ def self.exception
     assert_equal 3, @worker.processed
   end
 
+  it "reserve blocks when the queue is empty" do
+    worker = Resque::Worker.new(:timeout)
+
+    assert_raises Timeout::Error do
+      Timeout.timeout(1) { worker.reserve(5) }
+    end
+  end
+
   it "keeps track of how many failures it has seen" do
     Resque::Job.create(:jobs, BadJob)
     Resque::Job.create(:jobs, BadJob)