Merge pull request resque#666 from ecoffey/pause_hooks

Add before_pause and after_pause as Resque worker hooks

Author: steveklabnik

docs/HOOKS.md

@@ -38,6 +38,18 @@ The `after_fork` hook will be run in the child process and is passed
 the current job. Any changes you make, therefor, will only live as
 long as the job currently being processes.
 
+You can also run a hook both before a worker pauses (`before_pause`),
+and after it is paused (`after_pause`). In each case the block will be
+passed the worker that is pausing or unpausing, respectively:
+
+    Resque.before_pause do |worker|
+      puts "It looks like #{worker} is now paused!"
+    end
+
+    Resque.after_pause do |worker|
+      puts "Whew, back to work for #{worker}!"
+    end
+
 All worker hooks can also be set using a setter, e.g.
 
     Resque.after_fork = proc { puts "called" }

lib/resque.rb

@@ -121,6 +121,26 @@ def after_fork(&block)
   # Set the after_fork proc.
   attr_writer :after_fork
 
+  # The `before_pause` hook will be run in the parent process before the
+  # worker has paused processing (via #pause_processing or SIGUSR2).
+  def before_pause(&block)
+    @before_pause = block if block_given?
+    @before_pause
+  end
+
+  # Set the after_pause proc.
+  attr_writer :before_pause
+
+  # The `after_pause` hook will be run in the parent process after the
+  # worker has paused (via SIGCONT).
+  def after_pause(&block)
+    @after_pause = block if block_given?
+    @after_pause
+  end
+
+  # Set the after_continue proc.
+  attr_writer :after_pause
+
   def to_s
     "Resque Client connected to #{redis_id}"
   end

lib/resque/worker.rb

@@ -361,8 +361,10 @@ def pause
         wr.write 'x'
         wr.close
       }
+      run_hook :before_pause, self
       rd.read 1
       rd.close
+      run_hook :after_pause, self
     end
 
     # Stop processing jobs after the current one has completed (if we're

test/worker_test.rb

@@ -479,7 +479,7 @@ def self.perform
   it "returns PID of running process" do
     assert_equal @worker.to_s.split(":")[1].to_i, @worker.pid
   end
-  
+
   it "requeue failed queue" do
     queue = 'good_job'
     Resque::Failure.create(:exception => Exception.new, :worker => Resque::Worker.new(queue), :queue => queue, :payload => {'class' => 'GoodJob'})
@@ -506,6 +506,52 @@ def self.perform
     assert_not_equal original_connection, Resque.redis.client.connection.instance_variable_get("@sock")
   end
 
+  it "will call before_pause before it is paused" do
+    before_pause_called = false
+    captured_worker = nil
+
+    Resque.before_pause do |worker|
+      before_pause_called = true
+      captured_worker = worker
+    end
+
+    @worker.pause_processing
+
+    assert !before_pause_called
+
+    t = Thread.start { sleep(0.1); Process.kill('CONT', @worker.pid) }
+
+    @worker.work(0)
+
+    t.join
+
+    assert before_pause_called
+    assert_equal @worker, captured_worker
+  end
+
+  it "will call after_pause after it is paused" do
+    after_pause_called = false
+    captured_worker = nil
+
+    Resque.after_pause do |worker|
+      after_pause_called = true
+      captured_worker = worker
+    end
+
+    @worker.pause_processing
+
+    assert !after_pause_called
+
+    t = Thread.start { sleep(0.1); Process.kill('CONT', @worker.pid) }
+
+    @worker.work(0)
+
+    t.join
+
+    assert after_pause_called
+    assert_equal @worker, captured_worker
+  end
+
   if !defined?(RUBY_ENGINE) || defined?(RUBY_ENGINE) && RUBY_ENGINE != "jruby"
     [SignalException, Resque::TermException].each do |exception|
       {