resque / php-resque

lib/Resque.php

Indentation +42 added lines, -42 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.
@@ -28,9 +28,9 @@ 
 	 */
 	protected static $redisDatabase = 0;
 
-    /**
-     * @var string auth of Redis database
-     */
+	/**
+	 * @var string auth of Redis database
+	 */
 	protected static $auth;
 
 	/**
@@ -42,7 +42,7 @@ 
 	 *                      and returns a Resque_Redis instance, or
 	 *                      a nested array of servers with host/port pairs.
 	 * @param int $database
-     * @param string $auth
+	 * @param string $auth
 	 */
 	public static function setBackend($server, $database = 0, $auth = null)
 	{
@@ -70,8 +70,8 @@ 
 		}
 
 		if (!empty(self::$auth)) {
-            self::$redis->auth(self::$auth);
-        }
+			self::$redis->auth(self::$auth);
+		}
 
 		return self::$redis;
 	}
@@ -133,7 +133,7 @@ 
 	 */
 	public static function pop($queue)
 	{
-        $item = self::redis()->lpop('queue:' . $queue);
+		$item = self::redis()->lpop('queue:' . $queue);
 
 		if(!$item) {
 			return;
@@ -151,11 +151,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);
-	    }
+		}
 	}
 
 	/**
@@ -166,9 +166,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;
 	}
 
 	/**
@@ -181,28 +181,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)
-	    );
+		);
 	}
 
 	/**
@@ -330,7 +330,7 @@ 
 		while (!$finished) {
 			$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() .$originalQueue);
 			if (empty($string)) {
-			    $finished = true;
+				$finished = true;
 			}
 		}
 
@@ -353,29 +353,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;
 	}
 
 	/**
@@ -388,9 +388,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;
 	}
 
 	/*

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 (Resque::redis()->ping() === false) {
-			        throw new CredisException('redis ping() failed');
-			    }
+				if (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));

Braces +8 added lines, -16 removed lines

@@ -212,8 +212,7 @@ 
 					$this->logger->log(Psr\Log\LogLevel::INFO, 'Sleeping for {interval}', array('interval' => $interval));
 					if($this->paused) {
 						$this->updateProcLine('Paused');
-					}
-					else {
+					} else {
 						$this->updateProcLine('Waiting');
 					}
 
@@ -272,8 +271,7 @@ 
 					$job->fail(new Resque_Job_DirtyExitException(
 						'Job exited with exit code ' . $exitStatus
 					));
-				}
-				else
+				} else
 				{
 					if (in_array($job->getStatus(), array(Resque_Job_Status::STATUS_WAITING, Resque_Job_Status::STATUS_RUNNING)))
 					{
@@ -301,13 +299,11 @@ 
 		try {
 			Resque_Event::trigger('afterFork', $job);
 			$result = $job->perform();
-		}
-		catch(Exception $e) {
+		} catch(Exception $e) {
 			$this->logger->log(Psr\Log\LogLevel::CRITICAL, '{job} has failed {exception}', array('job' => $job, 'exception' => $e));
 			$job->fail($e);
 			return;
-		}
-		catch(Error $e) {
+		} catch(Error $e) {
 			$this->logger->log(Psr\Log\LogLevel::CRITICAL, '{job} has failed {exception}', array('job' => $job, 'exception' => $e));
 			$job->fail($e);
 			return;
@@ -405,8 +401,7 @@ 
 		$processTitle = static::$processPrefix . '-' . Resque::VERSION . ' (' . implode(',', $this->queues) . '): ' . $status;
 		if(function_exists('cli_set_process_title') && PHP_OS !== 'Darwin') {
 			cli_set_process_title($processTitle);
-		}
-		else if(function_exists('setproctitle')) {
+		} else if(function_exists('setproctitle')) {
 			setproctitle($processTitle);
 		}
 	}
@@ -496,8 +491,7 @@ 
 			$this->logger->log(Psr\Log\LogLevel::DEBUG, 'Child {child} found, killing.', array('child' => $this->child));
 			posix_kill($this->child, SIGKILL);
 			$this->child = null;
-		}
-		else {
+		} else {
 			$this->logger->log(Psr\Log\LogLevel::INFO, 'Child {child} not found, restarting.', array('child' => $this->child));
 			$this->shutdown();
 		}
@@ -542,8 +536,7 @@ 
 				$line = preg_replace('/\s+/m', ' ', $line);
 				list(,,$pids[]) = explode(' ', trim($line), 3);
 			}
-		}
-		else {
+		} else {
 			exec('ps -A -o pid,args | grep [r]esque', $cmdOutput);
 			foreach($cmdOutput as $line) {
 				list($pids[],) = explode(' ', trim($line), 2);
@@ -628,8 +621,7 @@ 
 		$job = Resque::redis()->get('worker:' . $this);
 		if(!$job) {
 			return array();
-		}
-		else {
+		} else {
 			return json_decode($job, true);
 		}
 	}

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;
 	}
 }

Braces +6 added lines, -12 removed lines

@@ -115,14 +115,11 @@ 
 		try {
 			if (is_object($client)) {
 				$this->driver = $client;
-			}
-			elseif (is_object($server)) {
+			} elseif (is_object($server)) {
 				$this->driver = $server;
-			}
-			elseif (is_array($server)) {
+			} elseif (is_array($server)) {
 				$this->driver = new Credis_Cluster($server);
-			}
-			else {
+			} else {
 				list($host, $port, $dsnDatabase, $user, $password, $options) = self::parseDsn($server);
 				// $user is not used, only $password
 
@@ -147,8 +144,7 @@ 
 			if ($database !== null) {
 				$this->driver->select($database);
 			}
-		}
-		catch(CredisException $e) {
+		} catch(CredisException $e) {
 			throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
 		}
 	}
@@ -243,15 +239,13 @@ 
 				foreach ($args[0] AS $i => $v) {
 					$args[0][$i] = self::$defaultNamespace . $v;
 				}
-			}
-			else {
+			} else {
 				$args[0] = self::$defaultNamespace . $args[0];
 			}
 		}
 		try {
 			return $this->driver->__call($name, $args);
-		}
-		catch (CredisException $e) {
+		} catch (CredisException $e) {
 			throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
 		}
 	}