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

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

lib/Resque.php

Spacing +16 added lines, -16 removed lines

@@ -87,7 +87,7 @@ 
 	 */
 	public static function fork()
 	{
-		if(!function_exists('pcntl_fork')) {
+		if (!function_exists('pcntl_fork')) {
 			return false;
 		}
 
@@ -96,7 +96,7 @@ 
 		self::$redis = null;
 
 		$pid = pcntl_fork();
-		if($pid === -1) {
+		if ($pid === -1) {
 			throw new RuntimeException('Unable to fork child worker.');
 		}
 
@@ -135,7 +135,7 @@ 
 	{
         $item = self::redis()->lpop('queue:' . $queue);
 
-		if(!$item) {
+		if (!$item) {
 			return;
 		}
 
@@ -151,7 +151,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);
@@ -182,13 +182,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;
 	    }
 
@@ -240,7 +240,7 @@ 
 		try {
 			Resque_Event::trigger('beforeEnqueue', $hookParams);
 		}
-		catch(Resque_Job_DontCreate $e) {
+		catch (Resque_Job_DontCreate $e) {
 			return false;
 		}
 
@@ -269,7 +269,7 @@ 
 	public static function queues()
 	{
 		$queues = self::redis()->smembers('queues');
-		if(!is_array($queues)) {
+		if (!is_array($queues)) {
 			$queues = array();
 		}
 		return $queues;
@@ -283,7 +283,7 @@ 
 	public static function items($queue, $start = 0, $stop = -1)
 	{
 		$list = self::redis()->lrange('queue:' . $queue, $start, $stop);
-		if(!is_array($list)) {
+		if (!is_array($list)) {
 			$list = array();
 		}
 		return $list;
@@ -304,9 +304,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;
@@ -314,7 +314,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 {
@@ -328,7 +328,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;
 			}
@@ -355,10 +355,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}]

lib/Resque/Job.php

Spacing +12 added lines, -12 removed lines

@@ -64,7 +64,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.'
 			);
@@ -77,7 +77,7 @@ 
 			'queue_time' => microtime(true),
 		));
 
-		if($monitor) {
+		if ($monitor) {
 			Resque_Job_Status::create($id, $prefix);
 		}
 
@@ -94,7 +94,7 @@ 
 	public static function reserve($queue)
 	{
 		$payload = Resque::pop($queue);
-		if(!is_array($payload)) {
+		if (!is_array($payload)) {
 			return false;
 		}
 
@@ -113,7 +113,7 @@ 
 	{
 		$item = Resque::blpop($queues, $timeout);
 
-		if(!is_array($item)) {
+		if (!is_array($item)) {
 			return false;
 		}
 
@@ -127,7 +127,7 @@ 
 	 */
 	public function updateStatus($status, $result = null)
 	{
-		if(empty($this->payload['id'])) {
+		if (empty($this->payload['id'])) {
 			return;
 		}
 
@@ -190,20 +190,20 @@ 
 			Resque_Event::trigger('beforePerform', $this);
 
 			$instance = $this->getInstance();
-			if(is_callable([$instance, 'setUp'])) {
+			if (is_callable([$instance, 'setUp'])) {
 				$instance->setUp();
 			}
 
 			$result = $instance->perform();
 
-			if(is_callable([$instance, 'tearDown'])) {
+			if (is_callable([$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) {
 			$result = false;
 		}
 
@@ -250,7 +250,7 @@ 
 	{
 		$status = new Resque_Job_Status($this->payload['id'], $this->payload['prefix']);
 		$monitor = false;
-		if($status->isTracking()) {
+		if ($status->isTracking()) {
 			$monitor = true;
 		}
 
@@ -265,13 +265,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 +40 added lines, -40 removed lines

@@ -1,5 +1,5 @@ 
 <?php
-declare(ticks = 1);
+declare(ticks=1);
 
 /**
  * Resque worker that handles checking queues for jobs, fetching them
@@ -76,14 +76,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);
     }
 
 	/**
@@ -102,12 +102,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;
@@ -132,7 +132,7 @@ 
 	 */
 	public static function find($workerId)
 	{
-		if(!self::exists($workerId) || false === strpos($workerId, ":")) {
+		if (!self::exists($workerId) || false === strpos($workerId, ":")) {
 			return false;
 		}
 
@@ -166,12 +166,12 @@ 
 		$this->updateProcLine('Starting');
 		$this->startup();
 
-        if(function_exists('pcntl_signal_dispatch')) {
+        if (function_exists('pcntl_signal_dispatch')) {
             pcntl_signal_dispatch();
         }
 
-		while(true) {
-			if($this->shutdown) {
+		while (true) {
+			if ($this->shutdown) {
 				break;
 			}
 
@@ -189,8 +189,8 @@ 
 
 			// 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 with blocking timeout ' . $interval);
 				} else {
@@ -200,17 +200,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 {
@@ -235,13 +235,13 @@ 
 				$this->updateProcLine($status);
 				$this->logger->log(Psr\Log\LogLevel::INFO, $status);
 
-				if(!empty($job->payload['id'])) {
+				if (!empty($job->payload['id'])) {
 					Resque_Job_PID::create($job->payload['id']);
 				}
 
 				$this->perform($job);
 
-				if(!empty($job->payload['id'])) {
+				if (!empty($job->payload['id'])) {
 					Resque_Job_PID::del($job->payload['id']);
 				}
 
@@ -250,7 +250,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);
@@ -258,7 +258,7 @@ 
 
 				// Wait until the child process finishes before continuing
 				while (pcntl_wait($status, WNOHANG) === 0) {
-					if(function_exists('pcntl_signal_dispatch')) {
+					if (function_exists('pcntl_signal_dispatch')) {
 			            pcntl_signal_dispatch();
 			        }
 
@@ -302,12 +302,12 @@ 
 			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;
@@ -331,26 +331,26 @@ 
 		}
 
 		$queues = $this->queues();
-		if(!is_array($queues)) {
+		if (!is_array($queues)) {
 			return;
 		}
 
-		if($blocking === true) {
-		    if(empty($queues)){
+		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;
             }
 			$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;
 				}
@@ -373,7 +373,7 @@ 
 	 */
 	public function queues($fetch = true)
 	{
-		if(!in_array('*', $this->queues) || $fetch == false) {
+		if (!in_array('*', $this->queues) || $fetch == false) {
 			return $this->queues;
 		}
 
@@ -403,10 +403,10 @@ 
 	private function updateProcLine($status)
 	{
 		$processTitle = static::$processPrefix . '-' . Resque::VERSION . ' (' . implode(',', $this->queues) . '): ' . $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);
 		}
 	}
@@ -421,7 +421,7 @@ 
 	 */
 	private function registerSigHandlers()
 	{
-		if(!function_exists('pcntl_signal')) {
+		if (!function_exists('pcntl_signal')) {
 			return;
 		}
 
@@ -486,13 +486,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,s -p ' . $this->child, $output, $returnCode) && $returnCode != 1) {
+		if (exec('ps -o pid,s -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;
@@ -515,10 +515,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));
@@ -538,14 +538,14 @@ 
 		$pids = array();
 		if (strtoupper(substr(PHP_OS, 0, 3)) === 'WIN') {
 			exec('WMIC path win32_process get Processid,Commandline | findstr resque | findstr /V findstr', $cmdOutput);
-			foreach($cmdOutput as $line) {
+			foreach ($cmdOutput as $line) {
 				$line = preg_replace('/\s+/m', ' ', $line);
-				list(,,$pids[]) = explode(' ', trim($line), 3);
+				list(,, $pids[]) = explode(' ', trim($line), 3);
 			}
 		}
 		else {
 			exec('ps -A -o pid,args | grep [r]esque', $cmdOutput);
-			foreach($cmdOutput as $line) {
+			foreach ($cmdOutput as $line) {
 				list($pids[],) = explode(' ', trim($line), 2);
 			}
 		}
@@ -566,7 +566,7 @@ 
 	 */
 	public function unregisterWorker()
 	{
-		if(is_object($this->currentJob)) {
+		if (is_object($this->currentJob)) {
 			$this->currentJob->fail(new Resque_Job_DirtyExitException);
 		}
 
@@ -626,7 +626,7 @@ 
 	public function job()
 	{
 		$job = Resque::redis()->get('worker:' . $this);
-		if(!$job) {
+		if (!$job) {
 			return array();
 		}
 		else {

lib/Resque/Job/Status.php

Spacing +11 added lines, -11 removed lines

@@ -63,7 +63,7 @@ 
 			'started' => time(),
 			'result'  => null,
 		);
-		Resque::redis()->set((string) $status, json_encode($statusPacket));
+		Resque::redis()->set((string)$status, json_encode($statusPacket));
 
 		return $status;
 	}
@@ -76,11 +76,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;
 		}
@@ -96,13 +96,13 @@ 
 	 */
 	public function update($status, $result = null)
 	{
-		$status = (int) $status;
+		$status = (int)$status;
 
-		if(!$this->isTracking()) {
+		if (!$this->isTracking()) {
 			return;
 		}
 
-		if($status < self::STATUS_WAITING || $status > self::STATUS_COMPLETE) {
+		if ($status < self::STATUS_WAITING || $status > self::STATUS_COMPLETE) {
 			return;
 		}
 
@@ -115,7 +115,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);
 		}
 	}
@@ -201,19 +201,19 @@ 
 	*/
 	protected function fetch($value = null)
 	{
-		if(!$this->isTracking()) {
+		if (!$this->isTracking()) {
 			return false;
 		}
 
 		$statusPacket = json_decode(Resque::redis()->get((string)$this), true);
-		if(!$statusPacket) {
+		if (!$statusPacket) {
 			return false;
 		}
 
-		if(empty($value)) {
+		if (empty($value)) {
 			return $statusPacket;
 		} else {
-			if(isset($statusPacket[$value])) {
+			if (isset($statusPacket[$value])) {
 				return $statusPacket[$value];
 			} else {
 				return null;

lib/ResqueScheduler.php

Spacing +6 added lines, -6 removed lines

@@ -108,14 +108,14 @@
      */
     public static function removeDelayed($queue, $class, $args)
     {
-       $destroyed=0;
-       $item=json_encode(self::jobToHash($queue, $class, $args));
-       $redis=Resque::redis();
+       $destroyed = 0;
+       $item = json_encode(self::jobToHash($queue, $class, $args));
+       $redis = Resque::redis();
 
-       foreach($redis->keys('delayed:*') as $key)
+       foreach ($redis->keys('delayed:*') as $key)
        {
-           $key=$redis->removePrefix($key);
-           $destroyed+=$redis->lrem($key,0,$item);
+           $key = $redis->removePrefix($key);
+           $destroyed += $redis->lrem($key, 0, $item);
        }
 
        return $destroyed;