resque / php-resque

demo/check_status.php

Spacing +5 added lines, -5 removed lines

@@ -1,5 +1,5 @@ 
 <?php
-if(empty($argv[1])) {
+if (empty($argv[1])) {
 	die('Specify the ID of a job to monitor the status of.');
 }
 
@@ -12,12 +12,12 @@ 
 //Resque::setBackend('redis://user:pass@a.host.name:3432/2');
 
 $status = new Resque_Job_Status($argv[1]);
-if(!$status->isTracking()) {
+if (!$status->isTracking()) {
 	die("Resque is not tracking the status of this job.\n");
 }
 
-echo "Tracking status of ".$argv[1].". Press [break] to stop.\n\n";
-while(true) {
-	fwrite(STDOUT, "Status of ".$argv[1]." is: ".$status->get()."\n");
+echo "Tracking status of " . $argv[1] . ". Press [break] to stop.\n\n";
+while (true) {
+	fwrite(STDOUT, "Status of " . $argv[1] . " is: " . $status->get() . "\n");
 	sleep(1);
 }
\ No newline at end of file

demo/queue.php

Spacing +2 added lines, -2 removed lines

@@ -1,5 +1,5 @@ 
 <?php
-if(empty($argv[1])) {
+if (empty($argv[1])) {
 	die('Specify the name of a job to add. e.g, php queue.php PHP_Job');
 }
 
@@ -23,4 +23,4 @@ 
         $jobId = Resque::enqueue($argv[1], $argv[2], $args, true);	
 }
 
-echo "Queued job ".$jobId."\n\n";
+echo "Queued job " . $jobId . "\n\n";

lib/Resque.php

Spacing +15 added lines, -15 removed lines

@@ -76,7 +76,7 @@ 
 	 */
 	public static function fork()
 	{
-		if(!function_exists('pcntl_fork')) {
+		if (!function_exists('pcntl_fork')) {
 			return false;
 		}
 
@@ -85,7 +85,7 @@ 
 		self::$redis = null;
 
 		$pid = pcntl_fork();
-		if($pid === -1) {
+		if ($pid === -1) {
 			throw new RuntimeException('Unable to fork child worker.');
 		}
 
@@ -124,7 +124,7 @@ 
 	{
         $item = self::redis()->lpop('queue:' . $queue);
 
-		if(!$item) {
+		if (!$item) {
 			return;
 		}
 
@@ -140,7 +140,7 @@ 
 	 */
 	public static function dequeue($queue, $items = Array())
 	{
-	    if(count($items) > 0) {
+	    if (count($items) > 0) {
 		return self::removeItems($queue, $items);
 	    } else {
 		return self::removeList($queue);
@@ -171,13 +171,13 @@ 
 	public static function blpop(array $queues, $timeout)
 	{
 	    $list = array();
-	    foreach($queues AS $queue) {
+	    foreach ($queues AS $queue) {
 		$list[] = 'queue:' . $queue;
 	    }
 
 	    $item = self::redis()->blpop($list, (int)$timeout);
 
-	    if(!$item) {
+	    if (!$item) {
 		return;
 	    }
 
@@ -228,7 +228,7 @@ 
 		try {
 			Resque_Event::trigger('beforeEnqueue', $hookParams);
 		}
-		catch(Resque_Job_DontCreate $e) {
+		catch (Resque_Job_DontCreate $e) {
 			return false;
 		}
 
@@ -257,7 +257,7 @@ 
 	public static function queues()
 	{
 		$queues = self::redis()->smembers('queues');
-		if(!is_array($queues)) {
+		if (!is_array($queues)) {
 			$queues = array();
 		}
 		return $queues;
@@ -278,9 +278,9 @@ 
 	private static function removeItems($queue, $items = Array())
 	{
 		$counter = 0;
-		$originalQueue = 'queue:'. $queue;
-		$tempQueue = $originalQueue. ':temp:'. time();
-		$requeueQueue = $tempQueue. ':requeue';
+		$originalQueue = 'queue:' . $queue;
+		$tempQueue = $originalQueue . ':temp:' . time();
+		$requeueQueue = $tempQueue . ':requeue';
 
 		// move each item from original queue to temp queue and process it
 		$finished = false;
@@ -288,7 +288,7 @@ 
 			$string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue);
 
 			if (!empty($string)) {
-				if(self::matchItem($string, $items)) {
+				if (self::matchItem($string, $items)) {
 					self::redis()->rpop($tempQueue);
 					$counter++;
 				} else {
@@ -302,7 +302,7 @@ 
 		// move back from temp queue to original queue
 		$finished = false;
 		while (!$finished) {
-			$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() .$originalQueue);
+			$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() . $originalQueue);
 			if (empty($string)) {
 			    $finished = true;
 			}
@@ -329,10 +329,10 @@ 
 	{
 	    $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}]

Braces +1 added lines, -2 removed lines

@@ -227,8 +227,7 @@
 		);
 		try {
 			Resque_Event::trigger('beforeEnqueue', $hookParams);
-		}
-		catch(Resque_Job_DontCreate $e) {
+		} catch(Resque_Job_DontCreate $e) {
 			return false;
 		}
 

lib/Resque/Job.php

Spacing +12 added lines, -12 removed lines

@@ -63,7 +63,7 @@ 
 			$id = Resque::generateJobId();
 		}
 
-		if($args !== null && !is_array($args)) {
+		if ($args !== null && !is_array($args)) {
 			throw new InvalidArgumentException(
 				'Supplied $args must be an array.'
 			);
@@ -75,7 +75,7 @@ 
 			'queue_time' => microtime(true),
 		));
 
-		if($monitor) {
+		if ($monitor) {
 			Resque_Job_Status::create($id);
 		}
 
@@ -92,7 +92,7 @@ 
 	public static function reserve($queue)
 	{
 		$payload = Resque::pop($queue);
-		if(!is_array($payload)) {
+		if (!is_array($payload)) {
 			return false;
 		}
 
@@ -111,7 +111,7 @@ 
 	{
 		$item = Resque::blpop($queues, $timeout);
 
-		if(!is_array($item)) {
+		if (!is_array($item)) {
 			return false;
 		}
 
@@ -125,7 +125,7 @@ 
 	 */
 	public function updateStatus($status)
 	{
-		if(empty($this->payload['id'])) {
+		if (empty($this->payload['id'])) {
 			return;
 		}
 
@@ -187,20 +187,20 @@ 
 			Resque_Event::trigger('beforePerform', $this);
 
 			$instance = $this->getInstance();
-			if(method_exists($instance, 'setUp')) {
+			if (method_exists($instance, 'setUp')) {
 				$instance->setUp();
 			}
 
 			$instance->perform();
 
-			if(method_exists($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) {
+		catch (Resque_Job_DontPerform $e) {
 			return false;
 		}
 
@@ -238,7 +238,7 @@ 
 	{
 		$status = new Resque_Job_Status($this->payload['id']);
 		$monitor = false;
-		if($status->isTracking()) {
+		if ($status->isTracking()) {
 			$monitor = true;
 		}
 
@@ -253,13 +253,13 @@ 
 	public function __toString()
 	{
 		$name = array(
-			'Job{' . $this->queue .'}'
+			'Job{' . $this->queue . '}'
 		);
-		if(!empty($this->payload['id'])) {
+		if (!empty($this->payload['id'])) {
 			$name[] = 'ID: ' . $this->payload['id'];
 		}
 		$name[] = $this->payload['class'];
-		if(!empty($this->payload['args'])) {
+		if (!empty($this->payload['args'])) {
 			$name[] = json_encode($this->payload['args']);
 		}
 		return '(' . implode(' | ', $name) . ')';

lib/Resque/Worker.php

Spacing +32 added lines, -32 removed lines

@@ -1,5 +1,5 @@ 
 <?php
-declare(ticks = 1);
+declare(ticks=1);
 
 /**
  * Resque worker that handles checking queues for jobs, fetching them
@@ -71,14 +71,14 @@ 
     {
         $this->logger = new Resque_Log();
 
-        if(!is_array($queues)) {
+        if (!is_array($queues)) {
             $queues = array($queues);
         }
 
         $this->queues = $queues;
         $this->hostname = php_uname('n');
 
-        $this->id = $this->hostname . ':'.getmypid() . ':' . implode(',', $this->queues);
+        $this->id = $this->hostname . ':' . getmypid() . ':' . implode(',', $this->queues);
     }
 
 	/**
@@ -88,12 +88,12 @@ 
 	public static function all()
 	{
 		$workers = Resque::redis()->smembers('workers');
-		if(!is_array($workers)) {
+		if (!is_array($workers)) {
 			$workers = array();
 		}
 
 		$instances = array();
-		foreach($workers as $workerId) {
+		foreach ($workers as $workerId) {
 			$instances[] = self::find($workerId);
 		}
 		return $instances;
@@ -118,7 +118,7 @@ 
 	 */
 	public static function find($workerId)
 	{
-		if(!self::exists($workerId) || false === strpos($workerId, ":")) {
+		if (!self::exists($workerId) || false === strpos($workerId, ":")) {
 			return false;
 		}
 
@@ -152,15 +152,15 @@ 
 		$this->updateProcLine('Starting');
 		$this->startup();
 
-		while(true) {
-			if($this->shutdown) {
+		while (true) {
+			if ($this->shutdown) {
 				break;
 			}
 
 			// Attempt to find and reserve a job
 			$job = false;
-			if(!$this->paused) {
-				if($blocking === true) {
+			if (!$this->paused) {
+				if ($blocking === true) {
 					$this->logger->log(Psr\Log\LogLevel::INFO, 'Starting blocking with timeout of {interval}', array('interval' => $interval));
 					$this->updateProcLine('Waiting for ' . implode(',', $this->queues) . ' with blocking timeout ' . $interval);
 				} else {
@@ -170,17 +170,17 @@ 
 				$job = $this->reserve($blocking, $interval);
 			}
 
-			if(!$job) {
+			if (!$job) {
 				// For an interval of 0, break now - helps with unit testing etc
-				if($interval == 0) {
+				if ($interval == 0) {
 					break;
 				}
 
-				if($blocking === false)
+				if ($blocking === false)
 				{
 					// If no job was found, we sleep for $interval before continuing and checking again
 					$this->logger->log(Psr\Log\LogLevel::INFO, 'Sleeping for {interval}', array('interval' => $interval));
-					if($this->paused) {
+					if ($this->paused) {
 						$this->updateProcLine('Paused');
 					}
 					else {
@@ -210,7 +210,7 @@ 
 				}
 			}
 
-			if($this->child > 0) {
+			if ($this->child > 0) {
 				// Parent process, sit and wait
 				$status = 'Forked ' . $this->child . ' at ' . strftime('%F %T');
 				$this->updateProcLine($status);
@@ -254,7 +254,7 @@ 
 			Resque_Event::trigger('afterFork', $job);
 			$job->perform();
 		}
-		catch(Exception $e) {
+		catch (Exception $e) {
 			$this->logger->log(Psr\Log\LogLevel::CRITICAL, '{job} has failed {stack}', array('job' => $job, 'stack' => $e));
 			$job->fail($e);
 			return;
@@ -278,21 +278,21 @@ 
 		}
 
 		$queues = $this->queues();
-		if(!is_array($queues)) {
+		if (!is_array($queues)) {
 			return;
 		}
 
-		if($blocking === true) {
+		if ($blocking === true) {
 			$job = Resque_Job::reserveBlocking($queues, $timeout);
-			if($job) {
+			if ($job) {
 				$this->logger->log(Psr\Log\LogLevel::INFO, 'Found job on {queue}', array('queue' => $job->queue));
 				return $job;
 			}
 		} else {
-			foreach($queues as $queue) {
+			foreach ($queues as $queue) {
 				$this->logger->log(Psr\Log\LogLevel::INFO, 'Checking {queue} for jobs', array('queue' => $queue));
 				$job = Resque_Job::reserve($queue);
-				if($job) {
+				if ($job) {
 					$this->logger->log(Psr\Log\LogLevel::INFO, 'Found job on {queue}', array('queue' => $job->queue));
 					return $job;
 				}
@@ -315,7 +315,7 @@ 
 	 */
 	public function queues($fetch = true)
 	{
-		if(!in_array('*', $this->queues) || $fetch == false) {
+		if (!in_array('*', $this->queues) || $fetch == false) {
 			return $this->queues;
 		}
 
@@ -345,10 +345,10 @@ 
 	private function updateProcLine($status)
 	{
 		$processTitle = 'resque-' . Resque::VERSION . ': ' . $status;
-		if(function_exists('cli_set_process_title') && PHP_OS !== 'Darwin') {
+		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);
 		}
 	}
@@ -363,7 +363,7 @@ 
 	 */
 	private function registerSigHandlers()
 	{
-		if(!function_exists('pcntl_signal')) {
+		if (!function_exists('pcntl_signal')) {
 			return;
 		}
 
@@ -421,13 +421,13 @@ 
 	 */
 	public function killChild()
 	{
-		if(!$this->child) {
+		if (!$this->child) {
 			$this->logger->log(Psr\Log\LogLevel::DEBUG, 'No child to kill.');
 			return;
 		}
 
 		$this->logger->log(Psr\Log\LogLevel::INFO, 'Killing child at {child}', array('child' => $this->child));
-		if(exec('ps -o pid,state -p ' . $this->child, $output, $returnCode) && $returnCode != 1) {
+		if (exec('ps -o pid,state -p ' . $this->child, $output, $returnCode) && $returnCode != 1) {
 			$this->logger->log(Psr\Log\LogLevel::DEBUG, 'Child {child} found, killing.', array('child' => $this->child));
 			posix_kill($this->child, SIGKILL);
 			$this->child = null;
@@ -450,10 +450,10 @@ 
 	{
 		$workerPids = $this->workerPids();
 		$workers = self::all();
-		foreach($workers as $worker) {
+		foreach ($workers as $worker) {
 			if (is_object($worker)) {
 				list($host, $pid, $queues) = explode(':', (string)$worker, 3);
-				if($host != $this->hostname || in_array($pid, $workerPids) || $pid == getmypid()) {
+				if ($host != $this->hostname || in_array($pid, $workerPids) || $pid == getmypid()) {
 					continue;
 				}
 				$this->logger->log(Psr\Log\LogLevel::INFO, 'Pruning dead worker: {worker}', array('worker' => (string)$worker));
@@ -472,7 +472,7 @@ 
 	{
 		$pids = array();
 		exec('ps -A -o pid,command | grep [r]esque', $cmdOutput);
-		foreach($cmdOutput as $line) {
+		foreach ($cmdOutput as $line) {
 			list($pids[],) = explode(' ', trim($line), 2);
 		}
 		return $pids;
@@ -492,7 +492,7 @@ 
 	 */
 	public function unregisterWorker()
 	{
-		if(is_object($this->currentJob)) {
+		if (is_object($this->currentJob)) {
 			$this->currentJob->fail(new Resque_Job_DirtyExitException);
 		}
 
@@ -552,7 +552,7 @@ 
 	public function job()
 	{
 		$job = Resque::redis()->get('worker:' . $this);
-		if(!$job) {
+		if (!$job) {
 			return array();
 		}
 		else {

Braces +6 added lines, -12 removed lines

@@ -182,8 +182,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 for ' . implode(',', $this->queues));
 					}
 
@@ -225,8 +224,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)))
 					{
@@ -253,8 +251,7 @@ 
 		try {
 			Resque_Event::trigger('afterFork', $job);
 			$job->perform();
-		}
-		catch(Exception $e) {
+		} catch(Exception $e) {
 			$this->logger->log(Psr\Log\LogLevel::CRITICAL, '{job} has failed {stack}', array('job' => $job, 'stack' => $e));
 			$job->fail($e);
 			return;
@@ -347,8 +344,7 @@ 
 		$processTitle = 'resque-' . Resque::VERSION . ': ' . $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);
 		}
 	}
@@ -431,8 +427,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();
 		}
@@ -554,8 +549,7 @@ 
 		$job = Resque::redis()->get('worker:' . $this);
 		if(!$job) {
 			return array();
-		}
-		else {
+		} else {
 			return json_decode($job, true);
 		}
 	}

lib/Resque/Job/Status.php

Spacing +6 added lines, -6 removed lines

@@ -66,11 +66,11 @@ 
 	 */
 	public function isTracking()
 	{
-		if($this->isTracking === false) {
+		if ($this->isTracking === false) {
 			return false;
 		}
 
-		if(!Resque::redis()->exists((string)$this)) {
+		if (!Resque::redis()->exists((string)$this)) {
 			$this->isTracking = false;
 			return false;
 		}
@@ -86,7 +86,7 @@ 
 	 */
 	public function update($status)
 	{
-		if(!$this->isTracking()) {
+		if (!$this->isTracking()) {
 			return;
 		}
 
@@ -97,7 +97,7 @@ 
 		Resque::redis()->set((string)$this, json_encode($statusPacket));
 
 		// Expire the status for completed jobs after 24 hours
-		if(in_array($status, self::$completeStatuses)) {
+		if (in_array($status, self::$completeStatuses)) {
 			Resque::redis()->expire((string)$this, 86400);
 		}
 	}
@@ -110,12 +110,12 @@ 
 	 */
 	public function get()
 	{
-		if(!$this->isTracking()) {
+		if (!$this->isTracking()) {
 			return false;
 		}
 
 		$statusPacket = json_decode(Resque::redis()->get((string)$this), true);
-		if(!$statusPacket) {
+		if (!$statusPacket) {
 			return false;
 		}
 

lib/Resque/Redis.php

Spacing +7 added lines, -7 removed lines

@@ -130,7 +130,7 @@ 
 
 				$this->driver = new Credis_Client($host, $port, $timeout, $persistent);
 				$this->driver->setMaxConnectRetries($maxRetries);
-				if ($password){
+				if ($password) {
 					$this->driver->auth($password);
 				}
 
@@ -145,7 +145,7 @@ 
 				$this->driver->select($database);
 			}
 		}
-		catch(CredisException $e) {
+		catch (CredisException $e) {
 			throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
 		}
 	}
@@ -170,7 +170,7 @@ 
 			// Use a sensible default for an empty DNS string
 			$dsn = 'redis://' . self::DEFAULT_HOST;
 		}
-		if(substr($dsn, 0, 7) === 'unix://') {
+		if (substr($dsn, 0, 7) === 'unix://') {
 			return array(
 				$dsn,
 				null,
@@ -184,12 +184,12 @@ 
 
 		// Check the URI scheme
 		$validSchemes = array('redis', 'tcp');
-		if (isset($parts['scheme']) && ! in_array($parts['scheme'], $validSchemes)) {
+		if (isset($parts['scheme']) && !in_array($parts['scheme'], $validSchemes)) {
 			throw new \InvalidArgumentException("Invalid DSN. Supported schemes are " . implode(', ', $validSchemes));
 		}
 
 		// Allow simple 'hostname' format, which `parse_url` treats as a path, not host.
-		if ( ! isset($parts['host']) && isset($parts['path'])) {
+		if (!isset($parts['host']) && isset($parts['path'])) {
 			$parts['host'] = $parts['path'];
 			unset($parts['path']);
 		}
@@ -260,10 +260,10 @@ 
 
 	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) );
+	        $string = substr($string, strlen($prefix), strlen($string));
 	    }
 	    return $string;
 	}

Braces +5 added lines, -10 removed lines

@@ -115,11 +115,9 @@ 
 		try {
 			if (is_array($server)) {
 				$this->driver = new Credis_Cluster($server);
-			}
-			else if (is_object($client)) {
+			} else if (is_object($client)) {
 				$this->driver = $client;
-			}
-			else {
+			} else {
 				list($host, $port, $dsnDatabase, $user, $password, $options) = self::parseDsn($server);
 				// $user is not used, only $password
 
@@ -144,8 +142,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);
 		}
 	}
@@ -240,15 +237,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);
 		}
 	}

lib/Resque/Failure.php

Spacing +1 added lines, -1 removed lines

@@ -35,7 +35,7 @@
 	 */
 	public static function getBackend()
 	{
-		if(self::$backend === null) {
+		if (self::$backend === null) {
 			self::$backend = 'Resque_Failure_Redis';
 		}