Add a plugin/event/hook system

Author: chris.boulton

README.markdown

@@ -227,6 +227,96 @@ A PECL module (<http://pecl.php.net/package/proctitle>) exists that
 adds this funcitonality to PHP, so if you'd like process titles updated,
 install the PECL module as well. php-resque will detect and use it.
 
+## Event/Hook System ##
+
+php-resque has a basic event system that can be used by your application
+to customize how some of the php-resque internals behave.
+
+You listen in on events (as listed below) by registering with `Resque_Event`
+and supplying a callback that you would like triggered when the event is
+raised:
+
+	Resque_Event::listen('eventName', [callback]);
+
+`[callback]` may be anything in PHP that is callable by `call_user_func_array`:
+
+* A string with the name of a function
+* An array containing an object and method to call
+* An array containing an object and a static method to call
+* A closure (PHP 5.3)
+
+Events may pass arguments (documented below), so your callback should accept
+these arguments.
+
+You can stop listening to an event by calling `Resque_Event::stopListening`
+with the same arguments supplied to `Resque_Event::listen`.
+
+It is up to your application to register event listeners. When enqueuing events
+in your application, it should be as easy as making sure php-resque is loaded
+and calling `Resque_Event::listen`.
+
+When running workers, if you run workers via the default `resque.php` script,
+your `APP_INCLUDE` script should initialize and register any listeners required
+for operation. If you have rolled your own worker manager, then it is again your
+responsibility to register listeners.
+
+A sample plugin is included in the `extras` directory.
+
+### Events ###
+
+#### beforeFirstFork ####
+
+Called once, as a worker initializes. Argument passed is the instance of `Resque_Worker`
+that was just initialized.
+
+#### beforeFork ####
+
+Called before php-resque forks to run a job. Argument passed contains the instance of
+`Resque_Job` for the job about to be run.
+
+`beforeFork` is triggered in the **parent** process. Any changes made will be permanent
+for as long as the worker lives.
+
+#### afterFork ####
+
+Called after php-resque forks to run a job (but before the job is run). Argument
+passed contains the instance of `Resque_Job` for the job about to be run.
+
+`afterFork` is triggered in the child process after forking out to complete a job. Any
+changes made will only live as long as the job is being processed.
+
+#### beforePerform ####
+
+Called before the `setUp` and `perform` methods on a job are run. Argument passed
+contains the instance of `Resque_Job` about for the job about to be run.
+
+You can prevent execution of the job by throwing an exception of `Resque_Job_DontPerform`.
+Any other exceptions thrown will be treated as if they were thrown in a job, causing the
+job to fail.
+
+#### afterPerform ####
+
+Called after the `perform` and `tearDown` methods on a job are run. Argument passed
+contains the instance of `Resque_Job` that was just run.
+
+Any exceptions thrown will be treated as if they were thrown in a job, causing the job
+to be marked as having failed.
+
+#### onFailure ####
+
+Called whenever a job fails. Arguments passed (in this order) include:
+
+* Exception - The exception that was thrown when the job failed
+* Resque_Job - The job that failed
+
+#### afterEnqueue ####
+
+Called after a job has been queued using the `Resque::enqueue` method. Arguments passed
+(in this order) include:
+
+* Class - string containing the name of the class the job was scheduled in
+* Arguments - array of arguments supplied to the job
+
 ## Contributors ##
 
 * chrisboulton

TODO.markdown

@@ -1,8 +1,6 @@
 * Write tests for:
   * `Resque_Failure`
   * `Resque_Failure_Redis`
-* Plugin/hook type system similar to Ruby version (when done, implement the
-setUp and tearDown methods as a plugin)
 * Change to preforking worker model
 * Clean up /bin and /demo
 * Add a way to store arbitrary text in job statuses (for things like progress

lib/Resque.php

@@ -1,4 +1,5 @@
 <?php
+require_once dirname(__FILE__) . '/Resque/Event.php';
 require_once dirname(__FILE__) . '/Resque/Exception.php';
 
 /**
@@ -103,7 +104,15 @@ public static function size($queue)
 	public static function enqueue($queue, $class, $args = null, $trackStatus = false)
 	{
 		require_once dirname(__FILE__) . '/Resque/Job.php';
-		return Resque_Job::create($queue, $class, $args, $trackStatus);
+		$result = Resque_Job::create($queue, $class, $args, $trackStatus);
+		if ($result) {
+			Resque_Event::trigger('afterEnqueue', array(
+				'class' => $class,
+				'args' => $args,
+			));
+		}
+		
+		return $result;
 	}
 
 	/**

lib/Resque/Event.php

@@ -0,0 +1,89 @@
+<?php
+/**
+ * Resque event/plugin system class
+ *
+ * @package		Resque/Event
+ * @author		Chris Boulton <[email protected]>
+ * @copyright	(c) 2010 Chris Boulton
+ * @license		http://www.opensource.org/licenses/mit-license.php
+ */
+class Resque_Event
+{
+	/**
+	 * @var array Array containing all registered callbacks, indexked by event name.
+	 */
+	private static $events = array();
+
+	/**
+	 * Raise a given event with the supplied data.
+	 *
+	 * @param string $event Name of event to be raised.
+	 * @param mixed $data Optional, any data that should be passed to each callback.
+	 * @return true
+	 */
+	public static function trigger($event, $data = null)
+	{
+		if (!is_array($data)) {
+			$data = array($data);
+		}
+
+		if (empty(self::$events[$event])) {
+			return true;
+		}
+		
+		foreach (self::$events[$event] as $callback) {
+			if (!is_callable($callback)) {
+				continue;
+			}
+			call_user_func_array($callback, $data);
+		}
+		
+		return true;
+	}
+	
+	/**
+	 * Listen in on a given event to have a specified callback fired.
+	 *
+	 * @param string $event Name of event to listen on.
+	 * @param mixed $callback Any callback callable by call_user_func_array.
+	 * @return true
+	 */
+	public static function listen($event, $callback)
+	{
+		if (!isset(self::$events[$event])) {
+			self::$events[$event] = array();
+		}
+		
+		self::$events[$event][] = $callback;
+		return true;
+	}
+	
+	/**
+	 * Stop a given callback from listening on a specific event.
+	 *
+	 * @param string $event Name of event.
+	 * @param mixed $callback The callback as defined when listen() was called.
+	 * @return true
+	 */
+	public static function stopListening($event, $callback)
+	{
+		if (!isset(self::$events[$event])) {
+			return true;
+		}
+		
+		$key = array_search($callback, self::$events[$event]);
+		if ($key !== false) {
+			unset(self::$events[$key]);
+		}
+		
+		return true;
+	}
+	
+	/**
+	 * Call all registered listeners.
+	 */
+	public static function clearListeners()
+	{
+		self::$events = array();
+	}
+}

lib/Resque/Job.php

@@ -1,5 +1,7 @@
 <?php
+require_once dirname(__FILE__) . '/Event.php';
 require_once dirname(__FILE__) . '/Job/Status.php';
+require_once dirname(__FILE__) . '/Job/DontPerform.php';
 
 /**
  * Resque job.
@@ -25,6 +27,11 @@ class Resque_Job
 	 * @var object Object containing details of the job.
 	 */
 	public $payload;
+	
+	/**
+	 * @var object Instance of the class performing work for this job.
+	 */
+	private $instance;
 
 	/**
 	 * Instantiate a new instance of a job.
@@ -109,15 +116,32 @@ public function getStatus()
 		$status = new Resque_Job_Status($this->payload['id']);
 		return $status->get();
 	}
-
+	
 	/**
-	 * Actually execute a job by calling the perform method on the class
-	 * associated with the job with the supplied arguments.
+	 * Get the arguments supplied to this job.
 	 *
-	 * @throws Resque_Exception When the job's class could not be found or it does not contain a perform method.
+	 * @return array Array of arguments.
 	 */
-	public function perform()
+	public function getArguments()
 	{
+		if (!isset($this->payload['args'])) {
+			return array();
+		}
+		
+		return $this->payload['args'];
+	}
+	
+	/**
+	 * Get the instantiated object for this job that will be performing work.
+	 *
+	 * @return object Instance of the object that this job belongs to.
+	 */
+	public function getInstance()
+	{
+		if (!is_null($this->instance)) {
+			return $this->instance;
+		}
+
 		if(!class_exists($this->payload['class'])) {
 			throw new Resque_Exception(
 				'Could not find job class ' . $this->payload['class'] . '.'
@@ -130,18 +154,51 @@ public function perform()
 			);
 		}
 
-		$instance = new $this->payload['class'];
-		$instance->args = $this->payload['args'];
+		$this->instance = new $this->payload['class'];
+		$this->instance->job = $this;
+		$this->instance->args = $this->getArguments();
+		return $this->instance;
+	}
 
-		if(method_exists($instance, 'setUp')) {
-			$instance->setUp();
-		}
+	/**
+	 * Actually execute a job by calling the perform method on the class
+	 * associated with the job with the supplied arguments.
+	 *
+	 * @throws Resque_Exception When the job's class could not be found or it does not contain a perform method.
+	 */
+	public function perform()
+	{
+		$instance = $this->getInstance();
+		try {
+			Resque_Event::trigger('beforePerform', $this);
+	
+			if(method_exists($instance, 'setUp')) {
+				$instance->setUp();
+			}
+
+			$instance->perform();
 
-		$instance->perform();
+			if(method_exists($instance, 'tearDown')) {
+				$instance->tearDown();
+			}
 
-		if(method_exists($instance, 'tearDown')) {
-			$instance->tearDown();
+			Resque_Event::trigger('afterPerform', $this);
+		}
+		// beforePerform/setUp have said don't perform this job. Return.
+		catch(Resque_Job_DontPerform $e) {
+			return false;
+		}
+		// Catch any other exception and raise the onFailure event. Rethrow
+		// the exception
+		catch(Exception $e) {
+			Resque_Event::trigger('onFailure', array(
+				'exception' => $e,
+				'job' => $this,
+			));
+			throw $e;
 		}
+		
+		return true;
 	}
 
 	/**

lib/Resque/Job/DontPerform.php

@@ -0,0 +1,13 @@
+<?php
+/**
+ * Exception to be thrown if a job should not be performed/run.
+ *
+ * @package		Resque/Job
+ * @author		Chris Boulton <[email protected]>
+ * @copyright	(c) 2010 Chris Boulton
+ * @license		http://www.opensource.org/licenses/mit-license.php
+ */
+class Resque_Job_DontPerform extends Exception
+{
+
+}

lib/Resque/Worker.php

@@ -1,5 +1,6 @@
 <?php
 require_once dirname(__FILE__) . '/Stat.php';
+require_once dirname(__FILE__) . '/Event.php';
 require_once dirname(__FILE__) . '/Job.php';
 require_once dirname(__FILE__) . '/Job/DirtyExitException.php';
 
@@ -180,11 +181,12 @@ public function work($interval = 5)
 				else {
 					$this->updateProcLine('Waiting for ' . implode(',', $this->queues));
 				}
-				usleep($interval*1000000);
+				usleep($interval * 1000000);
 				continue;
 			}
 
 			$this->log('got ' . $job);
+			Resque_Event::trigger('beforeFork', $job);
 			$this->workingOn($job);
 
 			$this->child = $this->fork();
@@ -231,6 +233,7 @@ public function work($interval = 5)
 	public function perform(Resque_Job $job)
 	{
 		try {
+			Resque_Event::trigger('afterFork', $job);
 			$job->perform();
 		}
 		catch(Exception $e) {
@@ -314,6 +317,7 @@ private function startup()
 	{
 		$this->registerSigHandlers();
 		$this->pruneDeadWorkers();
+		Resque_Event::trigger('beforeFirstFork');
 		$this->registerWorker();
 	}