resque / php-resque

demo/job.php

Indentation +1 added lines, -1 removed lines

@@ -3,7 +3,7 @@
 {
 	public function perform()
 	{
-        fwrite(STDOUT, 'Start job! -> ');
+		fwrite(STDOUT, 'Start job! -> ');
 		sleep(1);
 		fwrite(STDOUT, 'Job ended!' . PHP_EOL);
 	}

demo/queue.php

Indentation +1 added lines, -1 removed lines

@@ -20,7 +20,7 @@
 if (empty($argv[2])) {
 	$jobId = Resque::enqueue('default', $argv[1], $args, true);
 } else {
-        $jobId = Resque::enqueue($argv[1], $argv[2], $args, true);	
+		$jobId = Resque::enqueue($argv[1], $argv[2], $args, true);	
 }
 
 echo "Queued job ".$jobId."\n\n";

lib/Resque/Job/Factory.php

Indentation +24 added lines, -24 removed lines

@@ -3,30 +3,30 @@
 class Resque_Job_Factory implements Resque_Job_FactoryInterface
 {
 
-    /**
-     * @param $className
-     * @param array $args
-     * @param $queue
-     * @return Resque_JobInterface
-     * @throws \Resque_Exception
-     */
-    public function create($className, $args, $queue)
-    {
-        if (!class_exists($className)) {
-            throw new Resque_Exception(
-                'Could not find job class ' . $className . '.'
-            );
-        }
+	/**
+	 * @param $className
+	 * @param array $args
+	 * @param $queue
+	 * @return Resque_JobInterface
+	 * @throws \Resque_Exception
+	 */
+	public function create($className, $args, $queue)
+	{
+		if (!class_exists($className)) {
+			throw new Resque_Exception(
+				'Could not find job class ' . $className . '.'
+			);
+		}
 
-        if (!method_exists($className, 'perform')) {
-            throw new Resque_Exception(
-                'Job class ' . $className . ' does not contain a perform method.'
-            );
-        }
+		if (!method_exists($className, 'perform')) {
+			throw new Resque_Exception(
+				'Job class ' . $className . ' does not contain a perform method.'
+			);
+		}
 
-        $instance = new $className;
-        $instance->args = $args;
-        $instance->queue = $queue;
-        return $instance;
-    }
+		$instance = new $className;
+		$instance->args = $args;
+		$instance->queue = $queue;
+		return $instance;
+	}
 }

lib/Resque/Job/Status.php

Indentation +20 added lines, -20 removed lines

@@ -14,8 +14,8 @@ 
 	const STATUS_COMPLETE = 4;
 
 	/**
- 	 * @var string The prefix of the job status id.
- 	 */
+	 * @var string The prefix of the job status id.
+	 */
  	private $prefix;
 
 	/**
@@ -143,33 +143,33 @@ 
 	}
 
 	/**
- 	 * Fetch the last update timestamp of the job being monitored.
- 	 *
- 	 * @return mixed False if the job is not being monitored, otherwise the
+	 * Fetch the last update timestamp of the job being monitored.
+	 *
+	 * @return mixed False if the job is not being monitored, otherwise the
 	 *  update timestamp.
- 	 */
+	 */
 	public function updated()
 	{
 		return $this->fetch('updated');
  	}
 
 	/**
- 	 * Fetch the start timestamp of the job being monitored.
- 	 *
- 	 * @return mixed False if the job is not being monitored, otherwise the
+	 * Fetch the start timestamp of the job being monitored.
+	 *
+	 * @return mixed False if the job is not being monitored, otherwise the
 	 *  start timestamp.
- 	 */
+	 */
 	public function started()
 	{
 		return $this->fetch('started');
  	}
 
 	/**
- 	 * Fetch the result of the job being monitored.
- 	 *
- 	 * @return mixed False if the job is not being monitored, otherwise the result
- 	 * 	as mixed
- 	 */
+	 * Fetch the result of the job being monitored.
+	 *
+	 * @return mixed False if the job is not being monitored, otherwise the result
+	 * 	as mixed
+	 */
 	public function result()
 	{
 		return $this->fetch('result');
@@ -194,11 +194,11 @@ 
 	}
 
 	/**
-	* Fetch a value from the status packet for the job being monitored.
-	*
-	* @return mixed False if the status is not being monitored, otherwise the
-	*  requested value from the status packet.
-	*/
+	 * Fetch a value from the status packet for the job being monitored.
+	 *
+	 * @return mixed False if the status is not being monitored, otherwise the
+	 *  requested value from the status packet.
+	 */
 	protected function fetch($value = null)
 	{
 		if(!$this->isTracking()) {

lib/Resque/Redis.php

Indentation +11 added lines, -11 removed lines

@@ -98,10 +98,10 @@ 
 	 */
 	public static function prefix($namespace)
 	{
-	    if (substr($namespace, -1) !== ':' && $namespace != '') {
-	        $namespace .= ':';
-	    }
-	    self::$defaultNamespace = $namespace;
+		if (substr($namespace, -1) !== ':' && $namespace != '') {
+			$namespace .= ':';
+		}
+		self::$defaultNamespace = $namespace;
 	}
 
 	/**
@@ -110,7 +110,7 @@ 
 	 *                      DSN-supplied value will be used instead and this parameter is ignored.
 	 * @param object $client Optional Credis_Cluster or Credis_Client instance instantiated by you
 	 */
-    public function __construct($server, $database = null, $client = null)
+	public function __construct($server, $database = null, $client = null)
 	{
 		try {
 			if (is_object($client)) {
@@ -258,16 +258,16 @@ 
 
 	public static function getPrefix()
 	{
-	    return self::$defaultNamespace;
+		return self::$defaultNamespace;
 	}
 
 	public static function removePrefix($string)
 	{
-	    $prefix=self::getPrefix();
+		$prefix=self::getPrefix();
 
-	    if (substr($string, 0, strlen($prefix)) == $prefix) {
-	        $string = substr($string, strlen($prefix), strlen($string) );
-	    }
-	    return $string;
+		if (substr($string, 0, strlen($prefix)) == $prefix) {
+			$string = substr($string, strlen($prefix), strlen($string) );
+		}
+		return $string;
 	}
 }

lib/ResqueScheduler.php

Indentation +49 added lines, -49 removed lines

@@ -91,59 +91,59 @@
 		return Resque::redis()->llen('delayed:' . $timestamp, $timestamp);
 	}
 
-    /**
-     * Remove a delayed job from the queue
-     *
-     * note: you must specify exactly the same
-     * queue, class and arguments that you used when you added
-     * to the delayed queue
-     *
-     * also, this is an expensive operation because all delayed keys have tobe
-     * searched
-     *
-     * @param $queue
-     * @param $class
-     * @param $args
-     * @return int number of jobs that were removed
-     */
-    public static function removeDelayed($queue, $class, $args)
-    {
-       $destroyed=0;
-       $item=json_encode(self::jobToHash($queue, $class, $args));
-       $redis=Resque::redis();
+	/**
+	 * Remove a delayed job from the queue
+	 *
+	 * note: you must specify exactly the same
+	 * queue, class and arguments that you used when you added
+	 * to the delayed queue
+	 *
+	 * also, this is an expensive operation because all delayed keys have tobe
+	 * searched
+	 *
+	 * @param $queue
+	 * @param $class
+	 * @param $args
+	 * @return int number of jobs that were removed
+	 */
+	public static function removeDelayed($queue, $class, $args)
+	{
+	   $destroyed=0;
+	   $item=json_encode(self::jobToHash($queue, $class, $args));
+	   $redis=Resque::redis();
 
-       foreach($redis->keys('delayed:*') as $key)
-       {
-           $key=$redis->removePrefix($key);
-           $destroyed+=$redis->lrem($key,0,$item);
-       }
+	   foreach($redis->keys('delayed:*') as $key)
+	   {
+		   $key=$redis->removePrefix($key);
+		   $destroyed+=$redis->lrem($key,0,$item);
+	   }
 
-       return $destroyed;
-    }
+	   return $destroyed;
+	}
 
-    /**
-     * removed a delayed job queued for a specific timestamp
-     *
-     * note: you must specify exactly the same
-     * queue, class and arguments that you used when you added
-     * to the delayed queue
-     *
-     * @param $timestamp
-     * @param $queue
-     * @param $class
-     * @param $args
-     * @return mixed
-     */
-    public static function removeDelayedJobFromTimestamp($timestamp, $queue, $class, $args)
-    {
-        $key = 'delayed:' . self::getTimestamp($timestamp);
-        $item = json_encode(self::jobToHash($queue, $class, $args));
-        $redis = Resque::redis();
-        $count = $redis->lrem($key, 0, $item);
-        self::cleanupTimestamp($key, $timestamp);
+	/**
+	 * removed a delayed job queued for a specific timestamp
+	 *
+	 * note: you must specify exactly the same
+	 * queue, class and arguments that you used when you added
+	 * to the delayed queue
+	 *
+	 * @param $timestamp
+	 * @param $queue
+	 * @param $class
+	 * @param $args
+	 * @return mixed
+	 */
+	public static function removeDelayedJobFromTimestamp($timestamp, $queue, $class, $args)
+	{
+		$key = 'delayed:' . self::getTimestamp($timestamp);
+		$item = json_encode(self::jobToHash($queue, $class, $args));
+		$redis = Resque::redis();
+		$count = $redis->lrem($key, 0, $item);
+		self::cleanupTimestamp($key, $timestamp);
 
-        return $count;
-    }
+		return $count;
+	}
 	
 	/**
 	 * Generate hash of all job properties to be saved in the scheduled queue.

lib/ResqueScheduler/Worker.php

Indentation +7 added lines, -7 removed lines

@@ -36,13 +36,13 @@
 	private $paused = false;
 
 	/**
-	* The primary loop for a worker.
-	*
-	* Every $interval (seconds), the scheduled queue will be checked for jobs
-	* that should be pushed to Resque.
-	*
-	* @param int $interval How often to check schedules.
-	*/
+	 * The primary loop for a worker.
+	 *
+	 * Every $interval (seconds), the scheduled queue will be checked for jobs
+	 * that should be pushed to Resque.
+	 *
+	 * @param int $interval How often to check schedules.
+	 */
 	public function work($interval = null)
 	{
 		if ($interval !== null) {

lib/Resque/Worker.php

Indentation +40 added lines, -40 removed lines

@@ -17,8 +17,8 @@ 
 	private static $processPrefix = 'resque';
 
 	/**
-	* @var LoggerInterface Logging object that impliments the PSR-3 LoggerInterface
-	*/
+	 * @var LoggerInterface Logging object that impliments the PSR-3 LoggerInterface
+	 */
 	public $logger;
 
 	/**
@@ -61,30 +61,30 @@ 
 	 */
 	private $child = null;
 
-    /**
-     * Instantiate a new worker, given a list of queues that it should be working
-     * on. The list of queues should be supplied in the priority that they should
-     * be checked for jobs (first come, first served)
-     *
-     * Passing a single '*' allows the worker to work on all queues in alphabetical
-     * order. You can easily add new queues dynamically and have them worked on using
-     * this method.
-     *
-     * @param string|array $queues String with a single queue name, array with multiple.
-     */
-    public function __construct($queues)
-    {
-        $this->logger = new Resque_Log();
+	/**
+	 * Instantiate a new worker, given a list of queues that it should be working
+	 * on. The list of queues should be supplied in the priority that they should
+	 * be checked for jobs (first come, first served)
+	 *
+	 * Passing a single '*' allows the worker to work on all queues in alphabetical
+	 * order. You can easily add new queues dynamically and have them worked on using
+	 * this method.
+	 *
+	 * @param string|array $queues String with a single queue name, array with multiple.
+	 */
+	public function __construct($queues)
+	{
+		$this->logger = new Resque_Log();
 
-        if(!is_array($queues)) {
-            $queues = array($queues);
-        }
+		if(!is_array($queues)) {
+			$queues = array($queues);
+		}
 
-        $this->queues = $queues;
-        $this->hostname = php_uname('n');
+		$this->queues = $queues;
+		$this->hostname = php_uname('n');
 
-        $this->id = $this->hostname . ':'.getmypid() . ':' . implode(',', $this->queues);
-    }
+		$this->id = $this->hostname . ':'.getmypid() . ':' . implode(',', $this->queues);
+	}
 
 	/**
 	 * Set the process prefix of the workers to the given prefix string.
@@ -166,9 +166,9 @@ 
 		$this->updateProcLine('Starting');
 		$this->startup();
 
-        if(function_exists('pcntl_signal_dispatch')) {
-            pcntl_signal_dispatch();
-        }
+		if(function_exists('pcntl_signal_dispatch')) {
+			pcntl_signal_dispatch();
+		}
 
 		while(true) {
 			if($this->shutdown) {
@@ -177,14 +177,14 @@ 
 
 			// is redis still alive?
 			try {
-			    if (!$this->paused && Resque::redis()->ping() === false) {
-			        throw new CredisException('redis ping() failed');
-			    }
+				if (!$this->paused && Resque::redis()->ping() === false) {
+					throw new CredisException('redis ping() failed');
+				}
 			} catch (CredisException $e) {
-			    $this->logger->log(Psr\Log\LogLevel::ERROR, 'redis went away. trying to reconnect');
-			    Resque::$redis = null;
-			    usleep($interval * 1000000);
-			    continue;
+				$this->logger->log(Psr\Log\LogLevel::ERROR, 'redis went away. trying to reconnect');
+				Resque::$redis = null;
+				usleep($interval * 1000000);
+				continue;
 			}
 
 			// Attempt to find and reserve a job
@@ -259,8 +259,8 @@ 
 				// Wait until the child process finishes before continuing
 				while (pcntl_wait($status, WNOHANG) === 0) {
 					if(function_exists('pcntl_signal_dispatch')) {
-			            pcntl_signal_dispatch();
-			        }
+						pcntl_signal_dispatch();
+					}
 
 					// Pause for a half a second to conserve system resources
 					usleep(500000);
@@ -336,11 +336,11 @@ 
 		}
 
 		if($blocking === true) {
-		    if(empty($queues)){
-                $this->logger->log(Psr\Log\LogLevel::INFO, 'No queue was found, sleeping for {interval}', array('interval' => $timeout));
-                usleep($timeout * 1000000);
-                return false;
-            }
+			if(empty($queues)){
+				$this->logger->log(Psr\Log\LogLevel::INFO, 'No queue was found, sleeping for {interval}', array('interval' => $timeout));
+				usleep($timeout * 1000000);
+				return false;
+			}
 			$job = Resque_Job::reserveBlocking($queues, $timeout);
 			if($job) {
 				$this->logger->log(Psr\Log\LogLevel::INFO, 'Found job on {queue}', array('queue' => $job->queue));

lib/Resque.php

Indentation +44 added lines, -44 removed lines

@@ -10,7 +10,7 @@ 
 {
 	const VERSION = '1.2';
 
-    const DEFAULT_INTERVAL = 5;
+	const DEFAULT_INTERVAL = 5;
 
 	/**
 	 * @var Resque_Redis Instance of Resque_Redis that talks to redis or 
@@ -29,9 +29,9 @@ 
 	 */
 	protected static $redisDatabase = 0;
 
-    /**
-     * @var string auth of Redis database
-     */
+	/**
+	 * @var string auth of Redis database
+	 */
 	protected static $auth;
 
 	/**
@@ -44,7 +44,7 @@ 
 	 *                      a nested array of servers with host/port pairs or 
 	 *                      external Redis connection object.
 	 * @param int $database
-     * @param string $auth
+	 * @param string $auth
 	 */
 	public static function setBackend($server, $database = 0, $auth = null)
 	{
@@ -67,16 +67,16 @@ 
 		}
 
 		if (is_object(self::$redisServer)) {
-            		return self::$redisServer;
-        	} elseif (is_callable(self::$redisServer)) {
+					return self::$redisServer;
+			} elseif (is_callable(self::$redisServer)) {
 			self::$redis = call_user_func(self::$redisServer, self::$redisDatabase);
 		} else {
 			self::$redis = new Resque_Redis(self::$redisServer, self::$redisDatabase);
 		}
 
 		if (!empty(self::$auth)) {
-            self::$redis->auth(self::$auth);
-        }
+			self::$redis->auth(self::$auth);
+		}
 
 		return self::$redis;
 	}
@@ -138,7 +138,7 @@ 
 	 */
 	public static function pop($queue)
 	{
-        $item = self::redis()->lpop('queue:' . $queue);
+		$item = self::redis()->lpop('queue:' . $queue);
 
 		if(!$item) {
 			return;
@@ -156,11 +156,11 @@ 
 	 */
 	public static function dequeue($queue, $items = Array())
 	{
-	    if(count($items) > 0) {
+		if(count($items) > 0) {
 			return self::removeItems($queue, $items);
-	    } else {
+		} else {
 			return self::removeList($queue);
-	    }
+		}
 	}
 
 	/**
@@ -171,9 +171,9 @@ 
 	 */
 	public static function removeQueue($queue)
 	{
-	    $num = self::removeList($queue);
-	    self::redis()->srem('queues', $queue);
-	    return $num;
+		$num = self::removeList($queue);
+		self::redis()->srem('queues', $queue);
+		return $num;
 	}
 
 	/**
@@ -186,28 +186,28 @@ 
 	 */
 	public static function blpop(array $queues, $timeout)
 	{
-	    $list = array();
-	    foreach($queues AS $queue) {
+		$list = array();
+		foreach($queues AS $queue) {
 		$list[] = 'queue:' . $queue;
-	    }
+		}
 
-	    $item = self::redis()->blpop($list, (int)$timeout);
+		$item = self::redis()->blpop($list, (int)$timeout);
 
-	    if(!$item) {
+		if(!$item) {
 		return;
-	    }
+		}
 
-	    /**
-	     * Normally the Resque_Redis class returns queue names without the prefix
-	     * But the blpop is a bit different. It returns the name as prefix:queue:name
-	     * So we need to strip off the prefix:queue: part
-	     */
-	    $queue = substr($item[0], strlen(self::redis()->getPrefix() . 'queue:'));
+		/**
+		 * Normally the Resque_Redis class returns queue names without the prefix
+		 * But the blpop is a bit different. It returns the name as prefix:queue:name
+		 * So we need to strip off the prefix:queue: part
+		 */
+		$queue = substr($item[0], strlen(self::redis()->getPrefix() . 'queue:'));
 
-	    return array(
+		return array(
 		'queue'   => $queue,
 		'payload' => json_decode($item[1], true)
-	    );
+		);
 	}
 
 	/**
@@ -335,7 +335,7 @@ 
 		while (!$finished) {
 			$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() .$originalQueue);
 			if (empty($string)) {
-			    $finished = true;
+				$finished = true;
 			}
 		}
 
@@ -358,29 +358,29 @@ 
 	 */
 	private static function matchItem($string, $items)
 	{
-	    $decoded = json_decode($string, true);
+		$decoded = json_decode($string, true);
 
-	    foreach($items as $key => $val) {
+		foreach($items as $key => $val) {
 			# class name only  ex: item[0] = ['class']
 			if (is_numeric($key)) {
-			    if($decoded['class'] == $val) {
+				if($decoded['class'] == $val) {
 					return true;
-			    }
+				}
 			# class name with args , example: item[0] = ['class' => {'foo' => 1, 'bar' => 2}]
 			} elseif (is_array($val)) {
-			    $decodedArgs = (array)$decoded['args'][0];
-			    if ($decoded['class'] == $key &&
+				$decodedArgs = (array)$decoded['args'][0];
+				if ($decoded['class'] == $key &&
 					count($decodedArgs) > 0 && count(array_diff($decodedArgs, $val)) == 0) {
 					return true;
 				}
 			# class name with ID, example: item[0] = ['class' => 'id']
 			} else {
-			    if ($decoded['class'] == $key && $decoded['id'] == $val) {
+				if ($decoded['class'] == $key && $decoded['id'] == $val) {
 					return true;
-			    }
+				}
 			}
-	    }
-	    return false;
+		}
+		return false;
 	}
 
 	/**
@@ -393,9 +393,9 @@ 
 	 */
 	private static function removeList($queue)
 	{
-	    $counter = self::size($queue);
-	    $result = self::redis()->del('queue:' . $queue);
-	    return ($result == 1) ? $counter : 0;
+		$counter = self::size($queue);
+		$result = self::redis()->del('queue:' . $queue);
+		return ($result == 1) ? $counter : 0;
 	}
 
 	/*