idanoo/php-resque
1
0
Fork 0
mirror of https://github.com/idanoo/php-resque.git synced 2024-11-22 00:11:53 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

- Reformatted files to PSR2 standard

Browse Source 
- Removed credis for native phpredis
- Tidied up some docs
- Setting up new travis.ci build
This commit is contained in:
Daniel Mason 2018-05-25 19:26:54 +12:00
parent 065d5a4c63
commit ae84530132
30 changed files with 3724 additions and 2682 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGELOG.md
View File

@ -1,4 +1,8 @@
## 1.3 (2013-??-??) - Current Master ##
## 1.4 (2018-05-25)
## 1.3 (2013) ##
**Note:** This release introduces backwards incompatible changes with all previous versions of php-resque. Please see below for details.

4
composer.json
View File

@ -12,7 +12,7 @@
}
],
"require": {
"php": ">=5.3.0",
"php": ">=7.0.0",
"ext-pcntl": "*",
"psr/log": "~1.0"
},
@ -21,7 +21,7 @@
"ext-redis": "Native PHP extension for Redis connectivity. Credis will automatically utilize when available."
},
"require-dev": {
"phpunit/phpunit": "3.7.*"
"phpunit/phpunit": "^7"
},
"bin": [
"bin/resque"

1577
composer.lock generated
View File

File diff suppressed because it is too large Load Diff

661
lib/Resque.php
View File

@ -1,379 +1,380 @@
<?php
/**
* Base Resque class.
*
* @package Resque
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque
{
const VERSION = '1.2';
const VERSION = '1.4';
const DEFAULT_INTERVAL = 5;
/**
* @var Resque_Redis Instance of Resque_Redis that talks to redis.
*/
public static $redis = null;
/**
* @var Resque_Redis Instance of Resque_Redis that talks to redis.
*/
public static $redis = null;
/**
* @var mixed Host/port conbination separated by a colon, or a nested
* array of server swith host/port pairs
*/
protected static $redisServer = null;
/**
* @var mixed Host/port conbination separated by a colon, or a nested
* array of server swith host/port pairs
*/
protected static $redisServer = null;
/**
* @var int ID of Redis database to select.
*/
protected static $redisDatabase = 0;
/**
* @var int ID of Redis database to select.
*/
protected static $redisDatabase = 0;
/**
* Given a host/port combination separated by a colon, set it as
* the redis server that Resque will talk to.
*
* @param mixed $server Host/port combination separated by a colon, DSN-formatted URI, or
* a callable that receives the configured database ID
* and returns a Resque_Redis instance, or
* a nested array of servers with host/port pairs.
* @param int $database
*/
public static function setBackend($server, $database = 0)
{
self::$redisServer = $server;
self::$redisDatabase = $database;
self::$redis = null;
}
/**
* Given a host/port combination separated by a colon, set it as
* the redis server that Resque will talk to.
*
* @param mixed $server Host/port combination separated by a colon, DSN-formatted URI, or
* a callable that receives the configured database ID
* and returns a Resque_Redis instance, or
* a nested array of servers with host/port pairs.
* @param int $database
*/
public static function setBackend($server, $database = 0)
{
self::$redisServer = $server;
self::$redisDatabase = $database;
self::$redis = null;
}
/**
* Return an instance of the Resque_Redis class instantiated for Resque.
*
* @return Resque_Redis Instance of Resque_Redis.
*/
public static function redis()
{
if (self::$redis !== null) {
return self::$redis;
}
/**
* Return an instance of the Resque_Redis class instantiated for Resque.
*
* @return Resque_Redis Instance of Resque_Redis.
*/
public static function redis()
{
if (self::$redis !== null) {
return self::$redis;
}
if (is_callable(self::$redisServer)) {
self::$redis = call_user_func(self::$redisServer, self::$redisDatabase);
} else {
self::$redis = new Resque_Redis(self::$redisServer, self::$redisDatabase);
}
if (is_callable(self::$redisServer)) {
self::$redis = call_user_func(self::$redisServer, self::$redisDatabase);
} else {
self::$redis = new Resque_Redis(self::$redisServer, self::$redisDatabase);
}
return self::$redis;
}
return self::$redis;
}
/**
* fork() helper method for php-resque that handles issues PHP socket
* and phpredis have with passing around sockets between child/parent
* processes.
*
* Will close connection to Redis before forking.
*
* @return int Return vars as per pcntl_fork(). False if pcntl_fork is unavailable
*/
public static function fork()
{
if(!function_exists('pcntl_fork')) {
return false;
}
/**
* fork() helper method for php-resque that handles issues PHP socket
* and phpredis have with passing around sockets between child/parent
* processes.
*
* Will close connection to Redis before forking.
*
* @return int Return vars as per pcntl_fork(). False if pcntl_fork is unavailable
*/
public static function fork()
{
if (!function_exists('pcntl_fork')) {
return false;
}
// Close the connection to Redis before forking.
// This is a workaround for issues phpredis has.
self::$redis = null;
// Close the connection to Redis before forking.
// This is a workaround for issues phpredis has.
self::$redis = null;
$pid = pcntl_fork();
if($pid === -1) {
throw new RuntimeException('Unable to fork child worker.');
}
$pid = pcntl_fork();
if ($pid === -1) {
throw new RuntimeException('Unable to fork child worker.');
}
return $pid;
}
return $pid;
}
/**
* Push a job to the end of a specific queue. If the queue does not
* exist, then create it as well.
*
* @param string $queue The name of the queue to add the job to.
* @param array $item Job description as an array to be JSON encoded.
*/
public static function push($queue, $item)
{
$encodedItem = json_encode($item);
if ($encodedItem === false) {
return false;
}
self::redis()->sadd('queues', $queue);
$length = self::redis()->rpush('queue:' . $queue, $encodedItem);
if ($length < 1) {
return false;
}
return true;
}
/**
* Push a job to the end of a specific queue. If the queue does not
* exist, then create it as well.
*
* @param string $queue The name of the queue to add the job to.
* @param array $item Job description as an array to be JSON encoded.
*/
public static function push($queue, $item)
{
$encodedItem = json_encode($item);
if ($encodedItem === false) {
return false;
}
self::redis()->sadd('queues', $queue);
$length = self::redis()->rpush('queue:' . $queue, $encodedItem);
if ($length < 1) {
return false;
}
return true;
}
/**
* Pop an item off the end of the specified queue, decode it and
* return it.
*
* @param string $queue The name of the queue to fetch an item from.
* @return array Decoded item from the queue.
*/
public static function pop($queue)
{
/**
* Pop an item off the end of the specified queue, decode it and
* return it.
*
* @param string $queue The name of the queue to fetch an item from.
* @return array Decoded item from the queue.
*/
public static function pop($queue)
{
$item = self::redis()->lpop('queue:' . $queue);
if(!$item) {
return;
}
if (!$item) {
return;
}
return json_decode($item, true);
}
return json_decode($item, true);
}
/**
* Remove items of the specified queue
*
* @param string $queue The name of the queue to fetch an item from.
* @param array $items
* @return integer number of deleted items
*/
public static function dequeue($queue, $items = Array())
{
if(count($items) > 0) {
return self::removeItems($queue, $items);
} else {
return self::removeList($queue);
}
}
/**
* Remove items of the specified queue
*
* @param string $queue The name of the queue to fetch an item from.
* @param array $items
* @return integer number of deleted items
*/
public static function dequeue($queue, $items = Array())
{
if (count($items) > 0) {
return self::removeItems($queue, $items);
} else {
return self::removeList($queue);
}
}
/**
* Remove specified queue
*
* @param string $queue The name of the queue to remove.
* @return integer Number of deleted items
*/
public static function removeQueue($queue)
{
$num = self::removeList($queue);
self::redis()->srem('queues', $queue);
return $num;
}
/**
* Remove specified queue
*
* @param string $queue The name of the queue to remove.
* @return integer Number of deleted items
*/
public static function removeQueue($queue)
{
$num = self::removeList($queue);
self::redis()->srem('queues', $queue);
return $num;
}
/**
* Pop an item off the end of the specified queues, using blocking list pop,
* decode it and return it.
*
* @param array $queues
* @param int $timeout
* @return null|array Decoded item from the queue.
*/
public static function blpop(array $queues, $timeout)
{
$list = array();
foreach($queues AS $queue) {
$list[] = 'queue:' . $queue;
}
/**
* Pop an item off the end of the specified queues, using blocking list pop,
* decode it and return it.
*
* @param array $queues
* @param int $timeout
* @return null|array Decoded item from the queue.
*/
public static function blpop(array $queues, $timeout)
{
$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) {
return;
}
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(
'queue' => $queue,
'payload' => json_decode($item[1], true)
);
}
return array(
'queue' => $queue,
'payload' => json_decode($item[1], true)
);
}
/**
* Return the size (number of pending jobs) of the specified queue.
*
* @param string $queue name of the queue to be checked for pending jobs
*
* @return int The size of the queue.
*/
public static function size($queue)
{
return self::redis()->llen('queue:' . $queue);
}
/**
* Return the size (number of pending jobs) of the specified queue.
*
* @param string $queue name of the queue to be checked for pending jobs
*
* @return int The size of the queue.
*/
public static function size($queue)
{
return self::redis()->llen('queue:' . $queue);
}
/**
* Create a new job and save it to the specified queue.
*
* @param string $queue The name of the queue to place the job in.
* @param string $class The name of the class that contains the code to execute the job.
* @param array $args Any optional arguments that should be passed when the job is executed.
* @param boolean $trackStatus Set to true to be able to monitor the status of a job.
*
* @return string|boolean Job ID when the job was created, false if creation was cancelled due to beforeEnqueue
*/
public static function enqueue($queue, $class, $args = null, $trackStatus = false)
{
$id = Resque::generateJobId();
$hookParams = array(
'class' => $class,
'args' => $args,
'queue' => $queue,
'id' => $id,
);
try {
Resque_Event::trigger('beforeEnqueue', $hookParams);
}
catch(Resque_Job_DontCreate $e) {
return false;
}
/**
* Create a new job and save it to the specified queue.
*
* @param string $queue The name of the queue to place the job in.
* @param string $class The name of the class that contains the code to execute the job.
* @param array $args Any optional arguments that should be passed when the job is executed.
* @param boolean $trackStatus Set to true to be able to monitor the status of a job.
*
* @return string|boolean Job ID when the job was created, false if creation was cancelled due to beforeEnqueue
*/
public static function enqueue($queue, $class, $args = null, $trackStatus = false)
{
$id = Resque::generateJobId();
$hookParams = array(
'class' => $class,
'args' => $args,
'queue' => $queue,
'id' => $id,
);
try {
Resque_Event::trigger('beforeEnqueue', $hookParams);
} catch (Resque_Job_DontCreate $e) {
return false;
}
Resque_Job::create($queue, $class, $args, $trackStatus, $id);
Resque_Event::trigger('afterEnqueue', $hookParams);
Resque_Job::create($queue, $class, $args, $trackStatus, $id);
Resque_Event::trigger('afterEnqueue', $hookParams);
return $id;
}
return $id;
}
/**
* Reserve and return the next available job in the specified queue.
*
* @param string $queue Queue to fetch next available job from.
* @return Resque_Job Instance of Resque_Job to be processed, false if none or error.
*/
public static function reserve($queue)
{
return Resque_Job::reserve($queue);
}
/**
* Reserve and return the next available job in the specified queue.
*
* @param string $queue Queue to fetch next available job from.
* @return Resque_Job Instance of Resque_Job to be processed, false if none or error.
*/
public static function reserve($queue)
{
return Resque_Job::reserve($queue);
}
/**
* Get an array of all known queues.
*
* @return array Array of queues.
*/
public static function queues()
{
$queues = self::redis()->smembers('queues');
if(!is_array($queues)) {
$queues = array();
}
return $queues;
}
/**
* Get an array of all known queues.
*
* @return array Array of queues.
*/
public static function queues()
{
$queues = self::redis()->smembers('queues');
if (!is_array($queues)) {
$queues = array();
}
return $queues;
}
/**
* Remove Items from the queue
* Safely moving each item to a temporary queue before processing it
* If the Job matches, counts otherwise puts it in a requeue_queue
* which at the end eventually be copied back into the original queue
*
* @private
*
* @param string $queue The name of the queue
* @param array $items
* @return integer number of deleted items
*/
private static function removeItems($queue, $items = Array())
{
$counter = 0;
$originalQueue = 'queue:'. $queue;
$tempQueue = $originalQueue. ':temp:'. time();
$requeueQueue = $tempQueue. ':requeue';
/**
* Remove Items from the queue
* Safely moving each item to a temporary queue before processing it
* If the Job matches, counts otherwise puts it in a requeue_queue
* which at the end eventually be copied back into the original queue
*
* @private
*
* @param string $queue The name of the queue
* @param array $items
* @return integer number of deleted items
*/
private static function removeItems($queue, $items = Array())
{
$counter = 0;
$originalQueue = 'queue:' . $queue;
$tempQueue = $originalQueue . ':temp:' . time();
$requeueQueue = $tempQueue . ':requeue';
// move each item from original queue to temp queue and process it
$finished = false;
while (!$finished) {
$string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue);
// move each item from original queue to temp queue and process it
$finished = false;
while (!$finished) {
$string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue);
if (!empty($string)) {
if(self::matchItem($string, $items)) {
self::redis()->rpop($tempQueue);
$counter++;
} else {
self::redis()->rpoplpush($tempQueue, self::redis()->getPrefix() . $requeueQueue);
}
} else {
$finished = true;
}
}
if (!empty($string)) {
if (self::matchItem($string, $items)) {
self::redis()->rpop($tempQueue);
$counter++;
} else {
self::redis()->rpoplpush($tempQueue, self::redis()->getPrefix() . $requeueQueue);
}
} else {
$finished = true;
}
}
// move back from temp queue to original queue
$finished = false;
while (!$finished) {
$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() .$originalQueue);
if (empty($string)) {
$finished = true;
}
}
// move back from temp queue to original queue
$finished = false;
while (!$finished) {
$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() . $originalQueue);
if (empty($string)) {
$finished = true;
}
}
// remove temp queue and requeue queue
self::redis()->del($requeueQueue);
self::redis()->del($tempQueue);
// remove temp queue and requeue queue
self::redis()->del($requeueQueue);
self::redis()->del($tempQueue);
return $counter;
}
return $counter;
}
/**
* matching item
* item can be ['class'] or ['class' => 'id'] or ['class' => {:foo => 1, :bar => 2}]
* @private
*
* @params string $string redis result in json
* @params $items
*
* @return (bool)
*/
private static function matchItem($string, $items)
{
$decoded = json_decode($string, true);
/**
* matching item
* item can be ['class'] or ['class' => 'id'] or ['class' => {:foo => 1, :bar => 2}]
* @private
*
* @params string $string redis result in json
* @params $items
*
* @return (bool)
*/
private static function matchItem($string, $items)
{
$decoded = json_decode($string, true);
foreach($items as $key => $val) {
# class name only ex: item[0] = ['class']
if (is_numeric($key)) {
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 &&
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) {
return true;
}
}
}
return false;
}
foreach ($items as $key => $val) {
# class name only ex: item[0] = ['class']
if (is_numeric($key)) {
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 &&
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) {
return true;
}
}
}
return false;
}
/**
* Remove List
*
* @private
*
* @params string $queue the name of the queue
* @return integer number of deleted items belongs to this list
*/
private static function removeList($queue)
{
$counter = self::size($queue);
$result = self::redis()->del('queue:' . $queue);
return ($result == 1) ? $counter : 0;
}
/**
* Remove List
*
* @private
*
* @params string $queue the name of the queue
* @return integer number of deleted items belongs to this list
*/
private static function removeList($queue)
{
$counter = self::size($queue);
$result = self::redis()->del('queue:' . $queue);
return ($result == 1) ? $counter : 0;
}
/*
* Generate an identifier to attach to a job for status tracking.
*
* @return string
*/
public static function generateJobId()
{
return md5(uniqid('', true));
}
/*
* Generate an identifier to attach to a job for status tracking.
*
* @return string
*/
public static function generateJobId()
{
return md5(uniqid('', true));
}
}

158
lib/Resque/Event.php
View File

@ -1,88 +1,90 @@
<?php
/**
* Resque event/plugin system class
*
* @package Resque/Event
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Event
* @author Chris Boulton <chris@bigcommerce.com>
* @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();
/**
* @var array Array containing all registered callbacks, indexked by event name.
*/
private static $events = [];
/**
* 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);
}
/**
* 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 = [$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[$event][$key]);
}
return true;
}
/**
* Call all registered listeners.
*/
public static function clearListeners()
{
self::$events = array();
}
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] = [];
}
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[$event][$key]);
}
return true;
}
/**
* Call all registered listeners.
*/
public static function clearListeners()
{
self::$events = [];
}
}

8
lib/Resque/Exception.php
View File

@ -1,11 +1,13 @@
<?php
/**
* Resque exception.
*
* @package Resque
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Exception extends Exception
{
}

87
lib/Resque/Failure.php
View File

@ -3,54 +3,55 @@
/**
* Failed Resque job.
*
* @package Resque/Failure
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Failure
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Failure
{
/**
* @var string Class name representing the backend to pass failed jobs off to.
*/
private static $backend;
/**
* @var string Class name representing the backend to pass failed jobs off to.
*/
private static $backend;
/**
* Create a new failed job on the backend.
*
* @param object $payload The contents of the job that has just failed.
* @param \Exception $exception The exception generated when the job failed to run.
* @param \Resque_Worker $worker Instance of Resque_Worker that was running this job when it failed.
* @param string $queue The name of the queue that this job was fetched from.
*/
public static function create($payload, Exception $exception, Resque_Worker $worker, $queue)
{
$backend = self::getBackend();
new $backend($payload, $exception, $worker, $queue);
}
/**
* Create a new failed job on the backend.
*
* @param object $payload The contents of the job that has just failed.
* @param \Exception $exception The exception generated when the job failed to run.
* @param \Resque_Worker $worker Instance of Resque_Worker that was running this job when it failed.
* @param string $queue The name of the queue that this job was fetched from.
*/
public static function create($payload, Exception $exception, Resque_Worker $worker, $queue)
{
$backend = self::getBackend();
new $backend($payload, $exception, $worker, $queue);
}
/**
* Return an instance of the backend for saving job failures.
*
* @return object Instance of backend object.
*/
public static function getBackend()
{
if(self::$backend === null) {
self::$backend = 'Resque_Failure_Redis';
}
/**
* Return an instance of the backend for saving job failures.
*
* @return object Instance of backend object.
*/
public static function getBackend()
{
if (self::$backend === null) {
self::$backend = 'Resque_Failure_Redis';
}
return self::$backend;
}
return self::$backend;
}
/**
* Set the backend to use for raised job failures. The supplied backend
* should be the name of a class to be instantiated when a job fails.
* It is your responsibility to have the backend class loaded (or autoloaded)
*
* @param string $backend The class name of the backend to pipe failures to.
*/
public static function setBackend($backend)
{
self::$backend = $backend;
}
/**
* Set the backend to use for raised job failures. The supplied backend
* should be the name of a class to be instantiated when a job fails.
* It is your responsibility to have the backend class loaded (or autoloaded)
*
* @param string $backend The class name of the backend to pipe failures to.
*/
public static function setBackend($backend)
{
self::$backend = $backend;
}
}

25
lib/Resque/Failure/Interface.php
View File

@ -1,20 +1,21 @@
<?php
/**
* Interface that all failure backends should implement.
*
* @package Resque/Failure
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Failure
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
interface Resque_Failure_Interface
{
/**
* Initialize a failed job class and save it (where appropriate).
*
* @param object $payload Object containing details of the failed job.
* @param object $exception Instance of the exception that was thrown by the failed job.
* @param object $worker Instance of Resque_Worker that received the job.
* @param string $queue The name of the queue the job was fetched from.
*/
public function __construct($payload, $exception, $worker, $queue);
/**
* Initialize a failed job class and save it (where appropriate).
*
* @param object $payload Object containing details of the failed job.
* @param object $exception Instance of the exception that was thrown by the failed job.
* @param object $worker Instance of Resque_Worker that received the job.
* @param string $queue The name of the queue the job was fetched from.
*/
public function __construct($payload, $exception, $worker, $queue);
}

48
lib/Resque/Failure/Redis.php
View File

@ -2,32 +2,32 @@
/**
* Redis backend for storing failed Resque jobs.
*
* @package Resque/Failure
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Failure
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Failure_Redis implements Resque_Failure_Interface
{
/**
* Initialize a failed job class and save it (where appropriate).
*
* @param object $payload Object containing details of the failed job.
* @param object $exception Instance of the exception that was thrown by the failed job.
* @param object $worker Instance of Resque_Worker that received the job.
* @param string $queue The name of the queue the job was fetched from.
*/
public function __construct($payload, $exception, $worker, $queue)
{
$data = new stdClass;
$data->failed_at = strftime('%a %b %d %H:%M:%S %Z %Y');
$data->payload = $payload;
$data->exception = get_class($exception);
$data->error = $exception->getMessage();
$data->backtrace = explode("\n", $exception->getTraceAsString());
$data->worker = (string)$worker;
$data->queue = $queue;
$data = json_encode($data);
Resque::redis()->rpush('failed', $data);
}
/**
* Initialize a failed job class and save it (where appropriate).
*
* @param object $payload Object containing details of the failed job.
* @param object $exception Instance of the exception that was thrown by the failed job.
* @param object $worker Instance of Resque_Worker that received the job.
* @param string $queue The name of the queue the job was fetched from.
*/
public function __construct($payload, $exception, $worker, $queue)
{
$data = new stdClass;
$data->failed_at = strftime('%a %b %d %H:%M:%S %Z %Y');
$data->payload = $payload;
$data->exception = get_class($exception);
$data->error = $exception->getMessage();
$data->backtrace = explode("\n", $exception->getTraceAsString());
$data->worker = (string)$worker;
$data->queue = $queue;
$data = json_encode($data);
Resque::redis()->rpush('failed', $data);
}
}

2
lib/Resque/Job.php
View File

@ -1,4 +1,5 @@
<?php
/**
* Resque job.
*
@ -6,6 +7,7 @@
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Job
{
/**

9
lib/Resque/Job/DirtyExitException.php
View File

@ -1,12 +1,13 @@
<?php
/**
* Runtime exception class for a job that does not exit cleanly.
*
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Job_DirtyExitException extends RuntimeException
{
}
}

9
lib/Resque/Job/DontCreate.php
View File

@ -1,12 +1,13 @@
<?php
/**
* Exception to be thrown if while enqueuing a job it should not be created.
*
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Job_DontCreate extends Exception
{
}
}

10
lib/Resque/Job/DontPerform.php
View File

@ -1,12 +1,14 @@
<?php
/**
* Exception to be thrown if a job should not be performed/run.
*
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Job_DontPerform extends Exception
{
}
}

14
lib/Resque/Job/FactoryInterface.php
View File

@ -2,11 +2,11 @@
interface Resque_Job_FactoryInterface
{
/**
* @param $className
* @param array $args
* @param $queue
* @return Resque_JobInterface
*/
public function create($className, $args, $queue);
/**
* @param $className
* @param array $args
* @param $queue
* @return Resque_JobInterface
*/
public function create($className, $args, $queue);
}

238
lib/Resque/Job/Status.php
View File

@ -1,142 +1,144 @@
<?php
/**
* Status tracker/information for a job.
*
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Job_Status
{
const STATUS_WAITING = 1;
const STATUS_RUNNING = 2;
const STATUS_FAILED = 3;
const STATUS_COMPLETE = 4;
const STATUS_WAITING = 1;
const STATUS_RUNNING = 2;
const STATUS_FAILED = 3;
const STATUS_COMPLETE = 4;
/**
* @var string The ID of the job this status class refers back to.
*/
private $id;
/**
* @var string The ID of the job this status class refers back to.
*/
private $id;
/**
* @var mixed Cache variable if the status of this job is being monitored or not.
* True/false when checked at least once or null if not checked yet.
*/
private $isTracking = null;
/**
* @var mixed Cache variable if the status of this job is being monitored or not.
* True/false when checked at least once or null if not checked yet.
*/
private $isTracking = null;
/**
* @var array Array of statuses that are considered final/complete.
*/
private static $completeStatuses = array(
self::STATUS_FAILED,
self::STATUS_COMPLETE
);
/**
* @var array Array of statuses that are considered final/complete.
*/
private static $completeStatuses = array(
self::STATUS_FAILED,
self::STATUS_COMPLETE
);
/**
* Setup a new instance of the job monitor class for the supplied job ID.
*
* @param string $id The ID of the job to manage the status for.
*/
public function __construct($id)
{
$this->id = $id;
}
/**
* Setup a new instance of the job monitor class for the supplied job ID.
*
* @param string $id The ID of the job to manage the status for.
*/
public function __construct($id)
{
$this->id = $id;
}
/**
* Create a new status monitor item for the supplied job ID. Will create
* all necessary keys in Redis to monitor the status of a job.
*
* @param string $id The ID of the job to monitor the status of.
*/
public static function create($id)
{
$statusPacket = array(
'status' => self::STATUS_WAITING,
'updated' => time(),
'started' => time(),
);
Resque::redis()->set('job:' . $id . ':status', json_encode($statusPacket));
}
/**
* Create a new status monitor item for the supplied job ID. Will create
* all necessary keys in Redis to monitor the status of a job.
*
* @param string $id The ID of the job to monitor the status of.
*/
public static function create($id)
{
$statusPacket = array(
'status' => self::STATUS_WAITING,
'updated' => time(),
'started' => time(),
);
Resque::redis()->set('job:' . $id . ':status', json_encode($statusPacket));
}
/**
* Check if we're actually checking the status of the loaded job status
* instance.
*
* @return boolean True if the status is being monitored, false if not.
*/
public function isTracking()
{
if($this->isTracking === false) {
return false;
}
/**
* Check if we're actually checking the status of the loaded job status
* instance.
*
* @return boolean True if the status is being monitored, false if not.
*/
public function isTracking()
{
if ($this->isTracking === false) {
return false;
}
if(!Resque::redis()->exists((string)$this)) {
$this->isTracking = false;
return false;
}
if (!Resque::redis()->exists((string)$this)) {
$this->isTracking = false;
return false;
}
$this->isTracking = true;
return true;
}
$this->isTracking = true;
return true;
}
/**
* Update the status indicator for the current job with a new status.
*
* @param int The status of the job (see constants in Resque_Job_Status)
*/
public function update($status)
{
if(!$this->isTracking()) {
return;
}
/**
* Update the status indicator for the current job with a new status.
*
* @param int The status of the job (see constants in Resque_Job_Status)
*/
public function update($status)
{
if (!$this->isTracking()) {
return;
}
$statusPacket = array(
'status' => $status,
'updated' => time(),
);
Resque::redis()->set((string)$this, json_encode($statusPacket));
$statusPacket = [
'status' => $status,
'updated' => time(),
];
Resque::redis()->set((string)$this, json_encode($statusPacket));
// Expire the status for completed jobs after 24 hours
if(in_array($status, self::$completeStatuses)) {
Resque::redis()->expire((string)$this, 86400);
}
}
// Expire the status for completed jobs after 24 hours
if (in_array($status, self::$completeStatuses)) {
Resque::redis()->expire((string)$this, 86400);
}
}
/**
* Fetch the status for the job being monitored.
*
* @return mixed False if the status is not being monitored, otherwise the status as
* as an integer, based on the Resque_Job_Status constants.
*/
public function get()
{
if(!$this->isTracking()) {
return false;
}
/**
* Fetch the status for the job being monitored.
*
* @return mixed False if the status is not being monitored, otherwise the status as
* as an integer, based on the Resque_Job_Status constants.
*/
public function get()
{
if (!$this->isTracking()) {
return false;
}
$statusPacket = json_decode(Resque::redis()->get((string)$this), true);
if(!$statusPacket) {
return false;
}
$statusPacket = json_decode(Resque::redis()->get((string)$this), true);
if (!$statusPacket) {
return false;
}
return $statusPacket['status'];
}
return $statusPacket['status'];
}
/**
* Stop tracking the status of a job.
*/
public function stop()
{
Resque::redis()->del((string)$this);
}
/**
* Stop tracking the status of a job.
*/
public function stop()
{
Resque::redis()->del((string)$this);
}
/**
* Generate a string representation of this object.
*
* @return string String representation of the current job status class.
*/
public function __toString()
{
return 'job:' . $this->id . ':status';
}
/**
* Generate a string representation of this object.
*
* @return string String representation of the current job status class.
*/
public function __toString()
{
return 'job:' . $this->id . ':status';
}
}

8
lib/Resque/JobInterface.php
View File

@ -2,8 +2,8 @@
interface Resque_JobInterface
{
/**
* @return bool
*/
public function perform();
/**
* @return bool
*/
public function perform();
}

105
lib/Resque/Log.php
View File

@ -1,62 +1,65 @@
<?php
/**
* Resque default logger PSR-3 compliant
*
* @package Resque/Stat
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Stat
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Log extends Psr\Log\AbstractLogger
class Resque_Log extends Psr\Log\AbstractLogger
{
public $verbose;
public $verbose;
public function __construct($verbose = false) {
$this->verbose = $verbose;
}
public function __construct($verbose = false)
{
$this->verbose = $verbose;
}
/**
* Logs with an arbitrary level.
*
* @param mixed $level PSR-3 log level constant, or equivalent string
* @param string $message Message to log, may contain a { placeholder }
* @param array $context Variables to replace { placeholder }
* @return null
*/
public function log($level, $message, array $context = array())
{
if ($this->verbose) {
fwrite(
STDOUT,
'[' . $level . '] [' . strftime('%T %Y-%m-%d') . '] ' . $this->interpolate($message, $context) . PHP_EOL
);
return;
}
/**
* Logs with an arbitrary level.
*
* @param mixed $level PSR-3 log level constant, or equivalent string
* @param string $message Message to log, may contain a { placeholder }
* @param array $context Variables to replace { placeholder }
* @return null
*/
public function log($level, $message, array $context = array())
{
if ($this->verbose) {
fwrite(
STDOUT,
'[' . $level . '] [' . strftime('%T %Y-%m-%d') . '] ' . $this->interpolate($message, $context) . PHP_EOL
);
return;
}
if (!($level === Psr\Log\LogLevel::INFO || $level === Psr\Log\LogLevel::DEBUG)) {
fwrite(
STDOUT,
'[' . $level . '] ' . $this->interpolate($message, $context) . PHP_EOL
);
}
}
if (!($level === Psr\Log\LogLevel::INFO || $level === Psr\Log\LogLevel::DEBUG)) {
fwrite(
STDOUT,
'[' . $level . '] ' . $this->interpolate($message, $context) . PHP_EOL
);
}
}
/**
* Fill placeholders with the provided context
* @author Jordi Boggiano j.boggiano@seld.be
*
* @param string $message Message to be logged
* @param array $context Array of variables to use in message
* @return string
*/
public function interpolate($message, array $context = array())
{
// build a replacement array with braces around the context keys
$replace = array();
foreach ($context as $key => $val) {
$replace['{' . $key . '}'] = $val;
}
// interpolate replacement values into the message and return
return strtr($message, $replace);
}
/**
* Fill placeholders with the provided context
* @author Jordi Boggiano j.boggiano@seld.be
*
* @param string $message Message to be logged
* @param array $context Array of variables to use in message
* @return string
*/
public function interpolate($message, array $context = array())
{
// build a replacement array with braces around the context keys
$replace = array();
foreach ($context as $key => $val) {
$replace['{' . $key . '}'] = $val;
}
// interpolate replacement values into the message and return
return strtr($message, $replace);
}
}

472
lib/Resque/Redis.php
View File

@ -1,270 +1,262 @@
<?php
/**
* Wrap Credis to add namespace support and various helper methods.
* Set up phpredis connection
*
* @package Resque/Redis
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Redis
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Redis
{
/**
* Redis namespace
* @var string
*/
private static $defaultNamespace = 'resque:';
/**
* Redis namespace
* @var string
*/
private static $defaultNamespace = 'resque:';
/**
* A default host to connect to
*/
const DEFAULT_HOST = 'localhost';
/**
* A default host to connect to
*/
const DEFAULT_HOST = 'localhost';
/**
* The default Redis port
*/
const DEFAULT_PORT = 6379;
/**
* The default Redis port
*/
const DEFAULT_PORT = 6379;
/**
* The default Redis Database number
*/
const DEFAULT_DATABASE = 0;
/**
* The default Redis Database number
*/
const DEFAULT_DATABASE = 0;
/**
* @var array List of all commands in Redis that supply a key as their
* first argument. Used to prefix keys with the Resque namespace.
*/
private $keyCommands = array(
'exists',
'del',
'type',
'keys',
'expire',
'ttl',
'move',
'set',
'setex',
'get',
'getset',
'setnx',
'incr',
'incrby',
'decr',
'decrby',
'rpush',
'lpush',
'llen',
'lrange',
'ltrim',
'lindex',
'lset',
'lrem',
'lpop',
'blpop',
'rpop',
'sadd',
'srem',
'spop',
'scard',
'sismember',
'smembers',
'srandmember',
'zadd',
'zrem',
'zrange',
'zrevrange',
'zrangebyscore',
'zcard',
'zscore',
'zremrangebyscore',
'sort',
'rename',
'rpoplpush'
);
// sinterstore
// sunion
// sunionstore
// sdiff
// sdiffstore
// sinter
// smove
// mget
// msetnx
// mset
// renamenx
/**
* @var array List of all commands in Redis that supply a key as their
* first argument. Used to prefix keys with the Resque namespace.
*/
private $keyCommands = [
'exists',
'del',
'type',
'keys',
'expire',
'ttl',
'move',
'set',
'setex',
'get',
'getset',
'setnx',
'incr',
'incrby',
'decr',
'decrby',
'rpush',
'lpush',
'llen',
'lrange',
'ltrim',
'lindex',
'lset',
'lrem',
'lpop',
'blpop',
'rpop',
'sadd',
'srem',
'spop',
'scard',
'sismember',
'smembers',
'srandmember',
'zadd',
'zrem',
'zrange',
'zrevrange',
'zrangebyscore',
'zcard',
'zscore',
'zremrangebyscore',
'sort',
'rename',
'rpoplpush'
];
// sinterstore
// sunion
// sunionstore
// sdiff
// sdiffstore
// sinter
// smove
// mget
// msetnx
// mset
// renamenx
/**
* Set Redis namespace (prefix) default: resque
* @param string $namespace
*/
public static function prefix($namespace)
{
if (substr($namespace, -1) !== ':' && $namespace != '') {
$namespace .= ':';
}
self::$defaultNamespace = $namespace;
}
/**
* Set Redis namespace (prefix) default: resque
* @param string $namespace
*/
public static function prefix($namespace)
{
if (substr($namespace, -1) !== ':' && $namespace != '') {
$namespace .= ':';
}
self::$defaultNamespace = $namespace;
}
/**
* @param string|array $server A DSN or array
* @param int $database A database number to select. However, if we find a valid database number in the DSN the
* 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
*/
/**
* @param string|array $server A DSN or array
* @param int $database A database number to select. However, if we find a valid database number in the DSN the
* 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
* @throws Resque_RedisException
*/
public function __construct($server, $database = null, $client = null)
{
try {
if (is_array($server)) {
$this->driver = new Credis_Cluster($server);
}
else if (is_object($client)) {
$this->driver = $client;
}
else {
list($host, $port, $dsnDatabase, $user, $password, $options) = self::parseDsn($server);
// $user is not used, only $password
{
try {
if (is_object($client)) {
$this->redisConnection = $client;
} else {
list($host, $port, $dsnDatabase, $user, $password, $options) = self::parseDsn($server);
// $user is not used, only $password
// Look for known Credis_Client options
$timeout = isset($options['timeout']) ? intval($options['timeout']) : null;
$persistent = isset($options['persistent']) ? $options['persistent'] : '';
$maxRetries = isset($options['max_connect_retries']) ? $options['max_connect_retries'] : 0;
$timeout = isset($options['timeout']) ? intval($options['timeout']) : null;
$this->driver = new Credis_Client($host, $port, $timeout, $persistent);
$this->driver->setMaxConnectRetries($maxRetries);
if ($password){
$this->driver->auth($password);
}
$this->redisConnection = new \Redis();
// If we have found a database in our DSN, use it instead of the `$database`
// value passed into the constructor.
if ($dsnDatabase !== false) {
$database = $dsnDatabase;
}
}
if (!$this->redisConnection->connect($host, $port, $timeout)) {
throw new RedisException("Connection Failed to Redis!");
};
if ($database !== null) {
$this->driver->select($database);
}
}
catch(CredisException $e) {
throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
}
}
if ($password) {
$this->redisConnection->auth($password);
}
/**
* Parse a DSN string, which can have one of the following formats:
*
* - host:port
* - redis://user:pass@host:port/db?option1=val1&option2=val2
* - tcp://user:pass@host:port/db?option1=val1&option2=val2
* - unix:///path/to/redis.sock
*
* Note: the 'user' part of the DSN is not used.
*
* @param string $dsn A DSN string
* @return array An array of DSN compotnents, with 'false' values for any unknown components. e.g.
* [host, port, db, user, pass, options]
*/
public static function parseDsn($dsn)
{
if ($dsn == '') {
// Use a sensible default for an empty DNS string
$dsn = 'redis://' . self::DEFAULT_HOST;
}
if(substr($dsn, 0, 7) === 'unix://') {
return array(
$dsn,
null,
false,
null,
null,
null,
);
}
$parts = parse_url($dsn);
// If we have found a database in our DSN, use it instead of the `$database`
// value passed into the constructor.
if ($dsnDatabase !== false) {
$database = $dsnDatabase;
}
// Check the URI scheme
$validSchemes = array('redis', 'tcp');
if (isset($parts['scheme']) && ! in_array($parts['scheme'], $validSchemes)) {
throw new \InvalidArgumentException("Invalid DSN. Supported schemes are " . implode(', ', $validSchemes));
}
if ($database) {
$this->redisConnection->select($database);
}
}
} catch (RedisException $e) {
throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
}
}
// Allow simple 'hostname' format, which `parse_url` treats as a path, not host.
if ( ! isset($parts['host']) && isset($parts['path'])) {
$parts['host'] = $parts['path'];
unset($parts['path']);
}
/**
* Parse a DSN string, which can have one of the following formats:
*
* - host:port
* - redis://user:pass@host:port/db?option1=val1&option2=val2
* - tcp://user:pass@host:port/db?option1=val1&option2=val2
* - unix:///path/to/redis.sock
*
* Note: the 'user' part of the DSN is not used.
*
* @param string $dsn A DSN string
* @return array An array of DSN compotnents, with 'false' values for any unknown components. e.g.
* [host, port, db, user, pass, options]
*/
public static function parseDsn($dsn)
{
if ($dsn == '') {
// Use a sensible default for an empty DNS string
$dsn = 'redis://' . self::DEFAULT_HOST;
}
if (substr($dsn, 0, 7) === 'unix://') {
return [
$dsn,
null,
false,
null,
null,
null,
];
}
$parts = parse_url($dsn);
// Extract the port number as an integer
$port = isset($parts['port']) ? intval($parts['port']) : self::DEFAULT_PORT;
// Check the URI scheme
$validSchemes = array('redis', 'tcp');
if (isset($parts['scheme']) && !in_array($parts['scheme'], $validSchemes)) {
throw new \InvalidArgumentException("Invalid DSN. Supported schemes are " . implode(', ', $validSchemes));
}
// Get the database from the 'path' part of the URI
$database = false;
if (isset($parts['path'])) {
// Strip non-digit chars from path
$database = intval(preg_replace('/[^0-9]/', '', $parts['path']));
}
// Allow simple 'hostname' format, which `parse_url` treats as a path, not host.
if (!isset($parts['host']) && isset($parts['path'])) {
$parts['host'] = $parts['path'];
unset($parts['path']);
}
// Extract any 'user' and 'pass' values
$user = isset($parts['user']) ? $parts['user'] : false;
$pass = isset($parts['pass']) ? $parts['pass'] : false;
// Extract the port number as an integer
$port = isset($parts['port']) ? intval($parts['port']) : self::DEFAULT_PORT;
// Convert the query string into an associative array
$options = array();
if (isset($parts['query'])) {
// Parse the query string into an array
parse_str($parts['query'], $options);
}
// Get the database from the 'path' part of the URI
$database = false;
if (isset($parts['path'])) {
// Strip non-digit chars from path
$database = intval(preg_replace('/[^0-9]/', '', $parts['path']));
}
return array(
$parts['host'],
$port,
$database,
$user,
$pass,
$options,
);
}
// Extract any 'user' and 'pass' values
$user = isset($parts['user']) ? $parts['user'] : false;
$pass = isset($parts['pass']) ? $parts['pass'] : false;
/**
* Magic method to handle all function requests and prefix key based
* operations with the {self::$defaultNamespace} key prefix.
*
* @param string $name The name of the method called.
* @param array $args Array of supplied arguments to the method.
* @return mixed Return value from Resident::call() based on the command.
*/
public function __call($name, $args)
{
if (in_array($name, $this->keyCommands)) {
if (is_array($args[0])) {
foreach ($args[0] AS $i => $v) {
$args[0][$i] = self::$defaultNamespace . $v;
}
}
else {
$args[0] = self::$defaultNamespace . $args[0];
}
}
try {
return $this->driver->__call($name, $args);
}
catch (CredisException $e) {
throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
}
}
// Convert the query string into an associative array
$options = array();
if (isset($parts['query'])) {
// Parse the query string into an array
parse_str($parts['query'], $options);
}
public static function getPrefix()
{
return self::$defaultNamespace;
}
return array(
$parts['host'],
$port,
$database,
$user,
$pass,
$options,
);
}
public static function removePrefix($string)
{
$prefix=self::getPrefix();
/**
* Magic method to handle all function requests and prefix key based
* operations with the {self::$defaultNamespace} key prefix.
*
* @param string $name The name of the method called.
* @param array $args Array of supplied arguments to the method.
* @return mixed Return value from Resident::call() based on the command.
*/
public function __call($name, $args)
{
if (in_array($name, $this->keyCommands)) {
if (is_array($args[0])) {
foreach ($args[0] AS $i => $v) {
$args[0][$i] = self::$defaultNamespace . $v;
}
} else {
$args[0] = self::$defaultNamespace . $args[0];
}
}
}
if (substr($string, 0, strlen($prefix)) == $prefix) {
$string = substr($string, strlen($prefix), strlen($string) );
}
return $string;
}
public static function getPrefix()
{
return self::$defaultNamespace;
}
public static function removePrefix($string)
{
$prefix = self::getPrefix();
if (substr($string, 0, strlen($prefix)) == $prefix) {
$string = substr($string, strlen($prefix), strlen($string));
}
return $string;
}
}

10
lib/Resque/RedisException.php
View File

@ -1,12 +1,14 @@
<?php
/**
* Redis related exceptions
*
* @package Resque
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_RedisException extends Resque_Exception
{
}
?>

92
lib/Resque/Stat.php
View File

@ -1,56 +1,58 @@
<?php
/**
* Resque statistic management (jobs processed, failed, etc)
*
* @package Resque/Stat
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Stat
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Stat
{
/**
* Get the value of the supplied statistic counter for the specified statistic.
*
* @param string $stat The name of the statistic to get the stats for.
* @return mixed Value of the statistic.
*/
public static function get($stat)
{
return (int)Resque::redis()->get('stat:' . $stat);
}
/**
* Get the value of the supplied statistic counter for the specified statistic.
*
* @param string $stat The name of the statistic to get the stats for.
* @return mixed Value of the statistic.
*/
public static function get($stat)
{
return (int)Resque::redis()->get('stat:' . $stat);
}
/**
* Increment the value of the specified statistic by a certain amount (default is 1)
*
* @param string $stat The name of the statistic to increment.
* @param int $by The amount to increment the statistic by.
* @return boolean True if successful, false if not.
*/
public static function incr($stat, $by = 1)
{
return (bool)Resque::redis()->incrby('stat:' . $stat, $by);
}
/**
* Increment the value of the specified statistic by a certain amount (default is 1)
*
* @param string $stat The name of the statistic to increment.
* @param int $by The amount to increment the statistic by.
* @return boolean True if successful, false if not.
*/
public static function incr($stat, $by = 1)
{
return (bool)Resque::redis()->incrby('stat:' . $stat, $by);
}
/**
* Decrement the value of the specified statistic by a certain amount (default is 1)
*
* @param string $stat The name of the statistic to decrement.
* @param int $by The amount to decrement the statistic by.
* @return boolean True if successful, false if not.
*/
public static function decr($stat, $by = 1)
{
return (bool)Resque::redis()->decrby('stat:' . $stat, $by);
}
/**
* Decrement the value of the specified statistic by a certain amount (default is 1)
*
* @param string $stat The name of the statistic to decrement.
* @param int $by The amount to decrement the statistic by.
* @return boolean True if successful, false if not.
*/
public static function decr($stat, $by = 1)
{
return (bool)Resque::redis()->decrby('stat:' . $stat, $by);
}
/**
* Delete a statistic with the given name.
*
* @param string $stat The name of the statistic to delete.
* @return boolean True if successful, false if not.
*/
public static function clear($stat)
{
return (bool)Resque::redis()->del('stat:' . $stat);
}
/**
* Delete a statistic with the given name.
*
* @param string $stat The name of the statistic to delete.
* @return boolean True if successful, false if not.
*/
public static function clear($stat)
{
return (bool)Resque::redis()->del('stat:' . $stat);
}
}

941
lib/Resque/Worker.php
View File

File diff suppressed because it is too large Load Diff

320
test/Resque/Tests/EventTest.php
View File

@ -1,199 +1,201 @@
<?php
/**
* Resque_Event tests.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_EventTest extends Resque_Tests_TestCase
{
private $callbacksHit = array();
private $callbacksHit = array();
public function setUp()
{
Test_Job::$called = false;
public function setUp()
{
Test_Job::$called = false;
// Register a worker to test with
$this->worker = new Resque_Worker('jobs');
$this->worker->setLogger(new Resque_Log());
$this->worker->registerWorker();
}
// Register a worker to test with
$this->worker = new Resque_Worker('jobs');
$this->worker->setLogger(new Resque_Log());
$this->worker->registerWorker();
}
public function tearDown()
{
Resque_Event::clearListeners();
$this->callbacksHit = array();
}
public function tearDown()
{
Resque_Event::clearListeners();
$this->callbacksHit = array();
}
public function getEventTestJob()
{
$payload = array(
'class' => 'Test_Job',
'args' => array(
array('somevar'),
),
);
$job = new Resque_Job('jobs', $payload);
$job->worker = $this->worker;
return $job;
}
public function getEventTestJob()
{
$payload = array(
'class' => 'Test_Job',
'args' => array(
array('somevar'),
),
);
$job = new Resque_Job('jobs', $payload);
$job->worker = $this->worker;
return $job;
}
public function eventCallbackProvider()
{
return array(
array('beforePerform', 'beforePerformEventCallback'),
array('afterPerform', 'afterPerformEventCallback'),
array('afterFork', 'afterForkEventCallback'),
);
}
public function eventCallbackProvider()
{
return array(
array('beforePerform', 'beforePerformEventCallback'),
array('afterPerform', 'afterPerformEventCallback'),
array('afterFork', 'afterForkEventCallback'),
);
}
/**
* @dataProvider eventCallbackProvider
*/
public function testEventCallbacksFire($event, $callback)
{
Resque_Event::listen($event, array($this, $callback));
/**
* @dataProvider eventCallbackProvider
*/
public function testEventCallbacksFire($event, $callback)
{
Resque_Event::listen($event, array($this, $callback));
$job = $this->getEventTestJob();
$this->worker->perform($job);
$this->worker->work(0);
$job = $this->getEventTestJob();
$this->worker->perform($job);
$this->worker->work(0);
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback .') was not called');
}
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback . ') was not called');
}
public function testBeforeForkEventCallbackFires()
{
$event = 'beforeFork';
$callback = 'beforeForkEventCallback';
public function testBeforeForkEventCallbackFires()
{
$event = 'beforeFork';
$callback = 'beforeForkEventCallback';
Resque_Event::listen($event, array($this, $callback));
Resque::enqueue('jobs', 'Test_Job', array(
'somevar'
));
$job = $this->getEventTestJob();
$this->worker->work(0);
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback .') was not called');
}
Resque_Event::listen($event, array($this, $callback));
Resque::enqueue('jobs', 'Test_Job', array(
'somevar'
));
$job = $this->getEventTestJob();
$this->worker->work(0);
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback . ') was not called');
}
public function testBeforeEnqueueEventCallbackFires()
{
$event = 'beforeEnqueue';
$callback = 'beforeEnqueueEventCallback';
public function testBeforeEnqueueEventCallbackFires()
{
$event = 'beforeEnqueue';
$callback = 'beforeEnqueueEventCallback';
Resque_Event::listen($event, array($this, $callback));
Resque::enqueue('jobs', 'Test_Job', array(
'somevar'
));
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback .') was not called');
}
Resque_Event::listen($event, array($this, $callback));
Resque::enqueue('jobs', 'Test_Job', array(
'somevar'
));
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback . ') was not called');
}
public function testBeforePerformEventCanStopWork()
{
$callback = 'beforePerformEventDontPerformCallback';
Resque_Event::listen('beforePerform', array($this, $callback));
public function testBeforePerformEventCanStopWork()
{
$callback = 'beforePerformEventDontPerformCallback';
Resque_Event::listen('beforePerform', array($this, $callback));
$job = $this->getEventTestJob();
$job = $this->getEventTestJob();
$this->assertFalse($job->perform());
$this->assertContains($callback, $this->callbacksHit, $callback . ' callback was not called');
$this->assertFalse(Test_Job::$called, 'Job was still performed though Resque_Job_DontPerform was thrown');
}
$this->assertFalse($job->perform());
$this->assertContains($callback, $this->callbacksHit, $callback . ' callback was not called');
$this->assertFalse(Test_Job::$called, 'Job was still performed though Resque_Job_DontPerform was thrown');
}
public function testBeforeEnqueueEventStopsJobCreation()
{
$callback = 'beforeEnqueueEventDontCreateCallback';
Resque_Event::listen('beforeEnqueue', array($this, $callback));
Resque_Event::listen('afterEnqueue', array($this, 'afterEnqueueEventCallback'));
public function testBeforeEnqueueEventStopsJobCreation()
{
$callback = 'beforeEnqueueEventDontCreateCallback';
Resque_Event::listen('beforeEnqueue', array($this, $callback));
Resque_Event::listen('afterEnqueue', array($this, 'afterEnqueueEventCallback'));
$result = Resque::enqueue('test_job', 'TestClass');
$this->assertContains($callback, $this->callbacksHit, $callback . ' callback was not called');
$this->assertNotContains('afterEnqueueEventCallback', $this->callbacksHit, 'afterEnqueue was still called, even though it should not have been');
$this->assertFalse($result);
}
$result = Resque::enqueue('test_job', 'TestClass');
$this->assertContains($callback, $this->callbacksHit, $callback . ' callback was not called');
$this->assertNotContains('afterEnqueueEventCallback', $this->callbacksHit, 'afterEnqueue was still called, even though it should not have been');
$this->assertFalse($result);
}
public function testAfterEnqueueEventCallbackFires()
{
$callback = 'afterEnqueueEventCallback';
$event = 'afterEnqueue';
public function testAfterEnqueueEventCallbackFires()
{
$callback = 'afterEnqueueEventCallback';
$event = 'afterEnqueue';
Resque_Event::listen($event, array($this, $callback));
Resque::enqueue('jobs', 'Test_Job', array(
'somevar'
));
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback .') was not called');
}
Resque_Event::listen($event, array($this, $callback));
Resque::enqueue('jobs', 'Test_Job', array(
'somevar'
));
$this->assertContains($callback, $this->callbacksHit, $event . ' callback (' . $callback . ') was not called');
}
public function testStopListeningRemovesListener()
{
$callback = 'beforePerformEventCallback';
$event = 'beforePerform';
public function testStopListeningRemovesListener()
{
$callback = 'beforePerformEventCallback';
$event = 'beforePerform';
Resque_Event::listen($event, array($this, $callback));
Resque_Event::stopListening($event, array($this, $callback));
Resque_Event::listen($event, array($this, $callback));
Resque_Event::stopListening($event, array($this, $callback));
$job = $this->getEventTestJob();
$this->worker->perform($job);
$this->worker->work(0);
$job = $this->getEventTestJob();
$this->worker->perform($job);
$this->worker->work(0);
$this->assertNotContains($callback, $this->callbacksHit,
$event . ' callback (' . $callback .') was called though Resque_Event::stopListening was called'
);
}
$this->assertNotContains($callback, $this->callbacksHit,
$event . ' callback (' . $callback . ') was called though Resque_Event::stopListening was called'
);
}
public function beforePerformEventDontPerformCallback($instance)
{
$this->callbacksHit[] = __FUNCTION__;
throw new Resque_Job_DontPerform;
}
public function beforePerformEventDontPerformCallback($instance)
{
$this->callbacksHit[] = __FUNCTION__;
throw new Resque_Job_DontPerform;
}
public function beforeEnqueueEventDontCreateCallback($queue, $class, $args, $track = false)
{
$this->callbacksHit[] = __FUNCTION__;
throw new Resque_Job_DontCreate;
}
public function beforeEnqueueEventDontCreateCallback($queue, $class, $args, $track = false)
{
$this->callbacksHit[] = __FUNCTION__;
throw new Resque_Job_DontCreate;
}
public function assertValidEventCallback($function, $job)
{
$this->callbacksHit[] = $function;
if (!$job instanceof Resque_Job) {
$this->fail('Callback job argument is not an instance of Resque_Job');
}
$args = $job->getArguments();
$this->assertEquals($args[0], 'somevar');
}
public function assertValidEventCallback($function, $job)
{
$this->callbacksHit[] = $function;
if (!$job instanceof Resque_Job) {
$this->fail('Callback job argument is not an instance of Resque_Job');
}
$args = $job->getArguments();
$this->assertEquals($args[0], 'somevar');
}
public function afterEnqueueEventCallback($class, $args)
{
$this->callbacksHit[] = __FUNCTION__;
$this->assertEquals('Test_Job', $class);
$this->assertEquals(array(
'somevar',
), $args);
}
public function afterEnqueueEventCallback($class, $args)
{
$this->callbacksHit[] = __FUNCTION__;
$this->assertEquals('Test_Job', $class);
$this->assertEquals(array(
'somevar',
), $args);
}
public function beforeEnqueueEventCallback($job)
{
$this->callbacksHit[] = __FUNCTION__;
}
public function beforeEnqueueEventCallback($job)
{
$this->callbacksHit[] = __FUNCTION__;
}
public function beforePerformEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function beforePerformEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function afterPerformEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function afterPerformEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function beforeForkEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function beforeForkEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function afterForkEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
public function afterForkEventCallback($job)
{
$this->assertValidEventCallback(__FUNCTION__, $job);
}
}

160
test/Resque/Tests/JobStatusTest.php
View File

@ -1,11 +1,13 @@
<?php
/**
* Resque_Job_Status tests.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_JobStatusTest extends Resque_Tests_TestCase
{
/**
@ -13,94 +15,94 @@ class Resque_Tests_JobStatusTest extends Resque_Tests_TestCase
*/
protected $worker;
public function setUp()
{
parent::setUp();
public function setUp()
{
parent::setUp();
// Register a worker to test with
$this->worker = new Resque_Worker('jobs');
$this->worker->setLogger(new Resque_Log());
}
// Register a worker to test with
$this->worker = new Resque_Worker('jobs');
$this->worker->setLogger(new Resque_Log());
}
public function testJobStatusCanBeTracked()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$status = new Resque_Job_Status($token);
$this->assertTrue($status->isTracking());
}
public function testJobStatusCanBeTracked()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$status = new Resque_Job_Status($token);
$this->assertTrue($status->isTracking());
}
public function testJobStatusIsReturnedViaJobInstance()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$job = Resque_Job::reserve('jobs');
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $job->getStatus());
}
public function testJobStatusIsReturnedViaJobInstance()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$job = Resque_Job::reserve('jobs');
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $job->getStatus());
}
public function testQueuedJobReturnsQueuedStatus()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $status->get());
}
public function testQueuedJobReturnsQueuedStatus()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $status->get());
}
public function testRunningJobReturnsRunningStatus()
{
$token = Resque::enqueue('jobs', 'Failing_Job', null, true);
$job = $this->worker->reserve();
$this->worker->workingOn($job);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_RUNNING, $status->get());
}
public function testRunningJobReturnsRunningStatus()
{
$token = Resque::enqueue('jobs', 'Failing_Job', null, true);
$job = $this->worker->reserve();
$this->worker->workingOn($job);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_RUNNING, $status->get());
}
public function testFailedJobReturnsFailedStatus()
{
$token = Resque::enqueue('jobs', 'Failing_Job', null, true);
$this->worker->work(0);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_FAILED, $status->get());
}
public function testFailedJobReturnsFailedStatus()
{
$token = Resque::enqueue('jobs', 'Failing_Job', null, true);
$this->worker->work(0);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_FAILED, $status->get());
}
public function testCompletedJobReturnsCompletedStatus()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$this->worker->work(0);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_COMPLETE, $status->get());
}
public function testCompletedJobReturnsCompletedStatus()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, true);
$this->worker->work(0);
$status = new Resque_Job_Status($token);
$this->assertEquals(Resque_Job_Status::STATUS_COMPLETE, $status->get());
}
public function testStatusIsNotTrackedWhenToldNotTo()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, false);
$status = new Resque_Job_Status($token);
$this->assertFalse($status->isTracking());
}
public function testStatusIsNotTrackedWhenToldNotTo()
{
$token = Resque::enqueue('jobs', 'Test_Job', null, false);
$status = new Resque_Job_Status($token);
$this->assertFalse($status->isTracking());
}
public function testStatusTrackingCanBeStopped()
{
Resque_Job_Status::create('test');
$status = new Resque_Job_Status('test');
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $status->get());
$status->stop();
$this->assertFalse($status->get());
}
public function testStatusTrackingCanBeStopped()
{
Resque_Job_Status::create('test');
$status = new Resque_Job_Status('test');
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $status->get());
$status->stop();
$this->assertFalse($status->get());
}
public function testRecreatedJobWithTrackingStillTracksStatus()
{
$originalToken = Resque::enqueue('jobs', 'Test_Job', null, true);
$job = $this->worker->reserve();
public function testRecreatedJobWithTrackingStillTracksStatus()
{
$originalToken = Resque::enqueue('jobs', 'Test_Job', null, true);
$job = $this->worker->reserve();
// Mark this job as being worked on to ensure that the new status is still
// waiting.
$this->worker->workingOn($job);
// Mark this job as being worked on to ensure that the new status is still
// waiting.
$this->worker->workingOn($job);
// Now recreate it
$newToken = $job->recreate();
// Now recreate it
$newToken = $job->recreate();
// Make sure we've got a new job returned
$this->assertNotEquals($originalToken, $newToken);
// Make sure we've got a new job returned
$this->assertNotEquals($originalToken, $newToken);
// Now check the status of the new job
$newJob = Resque_Job::reserve('jobs');
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $newJob->getStatus());
}
// Now check the status of the new job
$newJob = Resque_Job::reserve('jobs');
$this->assertEquals(Resque_Job_Status::STATUS_WAITING, $newJob->getStatus());
}
}

770
test/Resque/Tests/JobTest.php
View File

@ -3,444 +3,446 @@
/**
* Resque_Job tests.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_JobTest extends Resque_Tests_TestCase
{
protected $worker;
protected $worker;
public function setUp()
{
parent::setUp();
public function setUp()
{
parent::setUp();
// Register a worker to test with
$this->worker = new Resque_Worker('jobs');
$this->worker->setLogger(new Resque_Log());
$this->worker->registerWorker();
}
// Register a worker to test with
$this->worker = new Resque_Worker('jobs');
$this->worker->setLogger(new Resque_Log());
$this->worker->registerWorker();
}
public function testJobCanBeQueued()
{
$this->assertTrue((bool)Resque::enqueue('jobs', 'Test_Job'));
}
public function testJobCanBeQueued()
{
$this->assertTrue((bool)Resque::enqueue('jobs', 'Test_Job'));
}
/**
* @expectedException Resque_RedisException
*/
public function testRedisErrorThrowsExceptionOnJobCreation()
{
$mockCredis = $this->getMockBuilder('Credis_Client')
->setMethods(['connect', '__call'])
->getMock();
$mockCredis->expects($this->any())->method('__call')
->will($this->throwException(new CredisException('failure')));
// public function testRedisErrorThrowsExceptionOnJobCreation()
// {
// $mockCredis = $this->getMockBuilder('Credis_Client')
// ->setMethods(['connect', '__call'])
// ->getMock();
// $mockCredis->expects($this->any())->method('__call')
// ->will($this->throwException(new CredisException('failure')));
//
// Resque::setBackend(function($database) use ($mockCredis) {
// return new Resque_Redis('localhost:6379', $database, $mockCredis);
// });
// Resque::enqueue('jobs', 'This is a test');
// }
Resque::setBackend(function($database) use ($mockCredis) {
return new Resque_Redis('localhost:6379', $database, $mockCredis);
});
Resque::enqueue('jobs', 'This is a test');
}
public function testQeueuedJobCanBeReserved()
{
Resque::enqueue('jobs', 'Test_Job');
public function testQeueuedJobCanBeReserved()
{
Resque::enqueue('jobs', 'Test_Job');
$job = Resque_Job::reserve('jobs');
if ($job == false) {
$this->fail('Job could not be reserved.');
}
$this->assertEquals('jobs', $job->queue);
$this->assertEquals('Test_Job', $job->payload['class']);
}
$job = Resque_Job::reserve('jobs');
if($job == false) {
$this->fail('Job could not be reserved.');
}
$this->assertEquals('jobs', $job->queue);
$this->assertEquals('Test_Job', $job->payload['class']);
}
/**
* @expectedException InvalidArgumentException
*/
public function testObjectArgumentsCannotBePassedToJob()
{
$args = new stdClass;
$args->test = 'somevalue';
Resque::enqueue('jobs', 'Test_Job', $args);
}
/**
* @expectedException InvalidArgumentException
*/
public function testObjectArgumentsCannotBePassedToJob()
{
$args = new stdClass;
$args->test = 'somevalue';
Resque::enqueue('jobs', 'Test_Job', $args);
}
public function testQueuedJobReturnsExactSamePassedInArguments()
{
$args = array(
'int' => 123,
'numArray' => array(
1,
2,
),
'assocArray' => array(
'key1' => 'value1',
'key2' => 'value2'
),
);
Resque::enqueue('jobs', 'Test_Job', $args);
$job = Resque_Job::reserve('jobs');
public function testQueuedJobReturnsExactSamePassedInArguments()
{
$args = array(
'int' => 123,
'numArray' => array(
1,
2,
),
'assocArray' => array(
'key1' => 'value1',
'key2' => 'value2'
),
);
Resque::enqueue('jobs', 'Test_Job', $args);
$job = Resque_Job::reserve('jobs');
$this->assertEquals($args, $job->getArguments());
}
$this->assertEquals($args, $job->getArguments());
}
public function testAfterJobIsReservedItIsRemoved()
{
Resque::enqueue('jobs', 'Test_Job');
Resque_Job::reserve('jobs');
$this->assertFalse(Resque_Job::reserve('jobs'));
}
public function testAfterJobIsReservedItIsRemoved()
{
Resque::enqueue('jobs', 'Test_Job');
Resque_Job::reserve('jobs');
$this->assertFalse(Resque_Job::reserve('jobs'));
}
public function testRecreatedJobMatchesExistingJob()
{
$args = array(
'int' => 123,
'numArray' => array(
1,
2,
),
'assocArray' => array(
'key1' => 'value1',
'key2' => 'value2'
),
);
public function testRecreatedJobMatchesExistingJob()
{
$args = array(
'int' => 123,
'numArray' => array(
1,
2,
),
'assocArray' => array(
'key1' => 'value1',
'key2' => 'value2'
),
);
Resque::enqueue('jobs', 'Test_Job', $args);
$job = Resque_Job::reserve('jobs');
Resque::enqueue('jobs', 'Test_Job', $args);
$job = Resque_Job::reserve('jobs');
// Now recreate it
$job->recreate();
// Now recreate it
$job->recreate();
$newJob = Resque_Job::reserve('jobs');
$this->assertEquals($job->payload['class'], $newJob->payload['class']);
$this->assertEquals($job->getArguments(), $newJob->getArguments());
}
$newJob = Resque_Job::reserve('jobs');
$this->assertEquals($job->payload['class'], $newJob->payload['class']);
$this->assertEquals($job->getArguments(), $newJob->getArguments());
}
public function testFailedJobExceptionsAreCaught()
{
$payload = array(
'class' => 'Failing_Job',
'args' => null
);
$job = new Resque_Job('jobs', $payload);
$job->worker = $this->worker;
public function testFailedJobExceptionsAreCaught()
{
$payload = array(
'class' => 'Failing_Job',
'args' => null
);
$job = new Resque_Job('jobs', $payload);
$job->worker = $this->worker;
$this->worker->perform($job);
$this->worker->perform($job);
$this->assertEquals(1, Resque_Stat::get('failed'));
$this->assertEquals(1, Resque_Stat::get('failed:'.$this->worker));
}
$this->assertEquals(1, Resque_Stat::get('failed'));
$this->assertEquals(1, Resque_Stat::get('failed:' . $this->worker));
}
/**
* @expectedException Resque_Exception
*/
public function testJobWithoutPerformMethodThrowsException()
{
Resque::enqueue('jobs', 'Test_Job_Without_Perform_Method');
$job = $this->worker->reserve();
$job->worker = $this->worker;
$job->perform();
}
/**
* @expectedException Resque_Exception
*/
public function testJobWithoutPerformMethodThrowsException()
{
Resque::enqueue('jobs', 'Test_Job_Without_Perform_Method');
$job = $this->worker->reserve();
$job->worker = $this->worker;
$job->perform();
}
/**
* @expectedException Resque_Exception
*/
public function testInvalidJobThrowsException()
{
Resque::enqueue('jobs', 'Invalid_Job');
$job = $this->worker->reserve();
$job->worker = $this->worker;
$job->perform();
}
public function testJobWithSetUpCallbackFiresSetUp()
{
$payload = array(
'class' => 'Test_Job_With_SetUp',
'args' => array(
'somevar',
'somevar2',
),
);
$job = new Resque_Job('jobs', $payload);
$job->perform();
$this->assertTrue(Test_Job_With_SetUp::$called);
}
public function testJobWithTearDownCallbackFiresTearDown()
{
$payload = array(
'class' => 'Test_Job_With_TearDown',
'args' => array(
'somevar',
'somevar2',
),
);
$job = new Resque_Job('jobs', $payload);
$job->perform();
$this->assertTrue(Test_Job_With_TearDown::$called);
}
/**
* @expectedException Resque_Exception
*/
public function testInvalidJobThrowsException()
{
Resque::enqueue('jobs', 'Invalid_Job');
$job = $this->worker->reserve();
$job->worker = $this->worker;
$job->perform();
}
public function testNamespaceNaming() {
$fixture = array(
array('test' => 'more:than:one:with:', 'assertValue' => 'more:than:one:with:'),
array('test' => 'more:than:one:without', 'assertValue' => 'more:than:one:without:'),
array('test' => 'resque', 'assertValue' => 'resque:'),
array('test' => 'resque:', 'assertValue' => 'resque:'),
);
public function testJobWithSetUpCallbackFiresSetUp()
{
$payload = array(
'class' => 'Test_Job_With_SetUp',
'args' => array(
'somevar',
'somevar2',
),
);
$job = new Resque_Job('jobs', $payload);
$job->perform();
foreach($fixture as $item) {
Resque_Redis::prefix($item['test']);
$this->assertEquals(Resque_Redis::getPrefix(), $item['assertValue']);
}
}
$this->assertTrue(Test_Job_With_SetUp::$called);
}
public function testJobWithNamespace()
{
Resque_Redis::prefix('php');
$queue = 'jobs';
$payload = array('another_value');
Resque::enqueue($queue, 'Test_Job_With_TearDown', $payload);
public function testJobWithTearDownCallbackFiresTearDown()
{
$payload = array(
'class' => 'Test_Job_With_TearDown',
'args' => array(
'somevar',
'somevar2',
),
);
$job = new Resque_Job('jobs', $payload);
$job->perform();
$this->assertEquals(Resque::queues(), array('jobs'));
$this->assertEquals(Resque::size($queue), 1);
$this->assertTrue(Test_Job_With_TearDown::$called);
}
Resque_Redis::prefix('resque');
$this->assertEquals(Resque::size($queue), 0);
}
public function testNamespaceNaming()
{
$fixture = array(
array('test' => 'more:than:one:with:', 'assertValue' => 'more:than:one:with:'),
array('test' => 'more:than:one:without', 'assertValue' => 'more:than:one:without:'),
array('test' => 'resque', 'assertValue' => 'resque:'),
array('test' => 'resque:', 'assertValue' => 'resque:'),
);
public function testDequeueAll()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$this->assertEquals(Resque::dequeue($queue), 2);
$this->assertEquals(Resque::size($queue), 0);
}
foreach ($fixture as $item) {
Resque_Redis::prefix($item['test']);
$this->assertEquals(Resque_Redis::getPrefix(), $item['assertValue']);
}
}
public function testDequeueMakeSureNotDeleteOthers()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue');
$other_queue = 'other_jobs';
Resque::enqueue($other_queue, 'Test_Job_Dequeue');
Resque::enqueue($other_queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$this->assertEquals(Resque::size($other_queue), 2);
$this->assertEquals(Resque::dequeue($queue), 2);
$this->assertEquals(Resque::size($queue), 0);
$this->assertEquals(Resque::size($other_queue), 2);
}
public function testJobWithNamespace()
{
Resque_Redis::prefix('php');
$queue = 'jobs';
$payload = array('another_value');
Resque::enqueue($queue, 'Test_Job_With_TearDown', $payload);
public function testDequeueSpecificItem()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue2');
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 1);
}
$this->assertEquals(Resque::queues(), array('jobs'));
$this->assertEquals(Resque::size($queue), 1);
public function testDequeueSpecificMultipleItems()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
Resque::enqueue($queue, 'Test_Job_Dequeue3');
$this->assertEquals(Resque::size($queue), 3);
$test = array('Test_Job_Dequeue2', 'Test_Job_Dequeue3');
$this->assertEquals(Resque::dequeue($queue, $test), 2);
$this->assertEquals(Resque::size($queue), 1);
}
Resque_Redis::prefix('resque');
$this->assertEquals(Resque::size($queue), 0);
}
public function testDequeueNonExistingItem()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
Resque::enqueue($queue, 'Test_Job_Dequeue3');
$this->assertEquals(Resque::size($queue), 3);
$test = array('Test_Job_Dequeue4');
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 3);
}
public function testDequeueAll()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$this->assertEquals(Resque::dequeue($queue), 2);
$this->assertEquals(Resque::size($queue), 0);
}
public function testDequeueNonExistingItem2()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
Resque::enqueue($queue, 'Test_Job_Dequeue3');
$this->assertEquals(Resque::size($queue), 3);
$test = array('Test_Job_Dequeue4', 'Test_Job_Dequeue1');
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 2);
}
public function testDequeueMakeSureNotDeleteOthers()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue');
$other_queue = 'other_jobs';
Resque::enqueue($other_queue, 'Test_Job_Dequeue');
Resque::enqueue($other_queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$this->assertEquals(Resque::size($other_queue), 2);
$this->assertEquals(Resque::dequeue($queue), 2);
$this->assertEquals(Resque::size($queue), 0);
$this->assertEquals(Resque::size($other_queue), 2);
}
public function testDequeueItemID()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
$qid = Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => $qid);
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueSpecificItem()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue2');
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueWrongItemID()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
$qid = Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
#qid right but class name is wrong
$test = array('Test_Job_Dequeue1' => $qid);
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 2);
}
public function testDequeueSpecificMultipleItems()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
Resque::enqueue($queue, 'Test_Job_Dequeue3');
$this->assertEquals(Resque::size($queue), 3);
$test = array('Test_Job_Dequeue2', 'Test_Job_Dequeue3');
$this->assertEquals(Resque::dequeue($queue, $test), 2);
$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueWrongItemID2()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
$qid = Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => 'r4nD0mH4sh3dId');
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 2);
}
public function testDequeueNonExistingItem()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
Resque::enqueue($queue, 'Test_Job_Dequeue3');
$this->assertEquals(Resque::size($queue), 3);
$test = array('Test_Job_Dequeue4');
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 3);
}
public function testDequeueItemWithArg()
{
$queue = 'jobs';
$arg = array('foo' => 1, 'bar' => 2);
Resque::enqueue($queue, 'Test_Job_Dequeue9');
Resque::enqueue($queue, 'Test_Job_Dequeue9', $arg);
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue9' => $arg);
$this->assertEquals(Resque::dequeue($queue, $test), 1);
#$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueSeveralItemsWithArgs()
{
// GIVEN
$queue = 'jobs';
$args = array('foo' => 1, 'bar' => 10);
$removeArgs = array('foo' => 1, 'bar' => 2);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $args);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $removeArgs);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $removeArgs);
$this->assertEquals(Resque::size($queue), 3);
// WHEN
$test = array('Test_Job_Dequeue9' => $removeArgs);
$removedItems = Resque::dequeue($queue, $test);
// THEN
$this->assertEquals($removedItems, 2);
$this->assertEquals(Resque::size($queue), 1);
$item = Resque::pop($queue);
$this->assertInternalType('array', $item['args']);
$this->assertEquals(10, $item['args'][0]['bar'], 'Wrong items were dequeued from queue!');
}
public function testDequeueNonExistingItem2()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue1');
Resque::enqueue($queue, 'Test_Job_Dequeue2');
Resque::enqueue($queue, 'Test_Job_Dequeue3');
$this->assertEquals(Resque::size($queue), 3);
$test = array('Test_Job_Dequeue4', 'Test_Job_Dequeue1');
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 2);
}
public function testDequeueItemWithUnorderedArg()
{
$queue = 'jobs';
$arg = array('foo' => 1, 'bar' => 2);
$arg2 = array('bar' => 2, 'foo' => 1);
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue', $arg);
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => $arg2);
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueItemID()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
$qid = Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => $qid);
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueItemWithiWrongArg()
{
$queue = 'jobs';
$arg = array('foo' => 1, 'bar' => 2);
$arg2 = array('foo' => 2, 'bar' => 3);
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue', $arg);
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => $arg2);
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 2);
}
public function testDequeueWrongItemID()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
$qid = Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
#qid right but class name is wrong
$test = array('Test_Job_Dequeue1' => $qid);
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 2);
}
public function testUseDefaultFactoryToGetJobInstance()
{
$payload = array(
'class' => 'Some_Job_Class',
'args' => null
);
$job = new Resque_Job('jobs', $payload);
$instance = $job->getInstance();
$this->assertInstanceOf('Some_Job_Class', $instance);
}
public function testDequeueWrongItemID2()
{
$queue = 'jobs';
Resque::enqueue($queue, 'Test_Job_Dequeue');
$qid = Resque::enqueue($queue, 'Test_Job_Dequeue');
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => 'r4nD0mH4sh3dId');
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 2);
}
public function testUseFactoryToGetJobInstance()
{
$payload = array(
'class' => 'Some_Job_Class',
'args' => array(array())
);
$job = new Resque_Job('jobs', $payload);
$factory = new Some_Stub_Factory();
$job->setJobFactory($factory);
$instance = $job->getInstance();
$this->assertInstanceOf('Resque_JobInterface', $instance);
}
public function testDequeueItemWithArg()
{
$queue = 'jobs';
$arg = array('foo' => 1, 'bar' => 2);
Resque::enqueue($queue, 'Test_Job_Dequeue9');
Resque::enqueue($queue, 'Test_Job_Dequeue9', $arg);
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue9' => $arg);
$this->assertEquals(Resque::dequeue($queue, $test), 1);
#$this->assertEquals(Resque::size($queue), 1);
}
public function testDoNotUseFactoryToGetInstance()
{
$payload = array(
'class' => 'Some_Job_Class',
'args' => array(array())
);
$job = new Resque_Job('jobs', $payload);
$factory = $this->getMock('Resque_Job_FactoryInterface');
$testJob = $this->getMock('Resque_JobInterface');
$factory->expects(self::never())->method('create')->will(self::returnValue($testJob));
$instance = $job->getInstance();
$this->assertInstanceOf('Resque_JobInterface', $instance);
}
public function testDequeueSeveralItemsWithArgs()
{
// GIVEN
$queue = 'jobs';
$args = array('foo' => 1, 'bar' => 10);
$removeArgs = array('foo' => 1, 'bar' => 2);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $args);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $removeArgs);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $removeArgs);
$this->assertEquals(Resque::size($queue), 3);
// WHEN
$test = array('Test_Job_Dequeue9' => $removeArgs);
$removedItems = Resque::dequeue($queue, $test);
// THEN
$this->assertEquals($removedItems, 2);
$this->assertEquals(Resque::size($queue), 1);
$item = Resque::pop($queue);
$this->assertInternalType('array', $item['args']);
$this->assertEquals(10, $item['args'][0]['bar'], 'Wrong items were dequeued from queue!');
}
public function testDequeueItemWithUnorderedArg()
{
$queue = 'jobs';
$arg = array('foo' => 1, 'bar' => 2);
$arg2 = array('bar' => 2, 'foo' => 1);
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue', $arg);
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => $arg2);
$this->assertEquals(Resque::dequeue($queue, $test), 1);
$this->assertEquals(Resque::size($queue), 1);
}
public function testDequeueItemWithiWrongArg()
{
$queue = 'jobs';
$arg = array('foo' => 1, 'bar' => 2);
$arg2 = array('foo' => 2, 'bar' => 3);
Resque::enqueue($queue, 'Test_Job_Dequeue');
Resque::enqueue($queue, 'Test_Job_Dequeue', $arg);
$this->assertEquals(Resque::size($queue), 2);
$test = array('Test_Job_Dequeue' => $arg2);
$this->assertEquals(Resque::dequeue($queue, $test), 0);
$this->assertEquals(Resque::size($queue), 2);
}
public function testUseDefaultFactoryToGetJobInstance()
{
$payload = array(
'class' => 'Some_Job_Class',
'args' => null
);
$job = new Resque_Job('jobs', $payload);
$instance = $job->getInstance();
$this->assertInstanceOf('Some_Job_Class', $instance);
}
public function testUseFactoryToGetJobInstance()
{
$payload = array(
'class' => 'Some_Job_Class',
'args' => array(array())
);
$job = new Resque_Job('jobs', $payload);
$factory = new Some_Stub_Factory();
$job->setJobFactory($factory);
$instance = $job->getInstance();
$this->assertInstanceOf('Resque_JobInterface', $instance);
}
public function testDoNotUseFactoryToGetInstance()
{
$payload = array(
'class' => 'Some_Job_Class',
'args' => array(array())
);
$job = new Resque_Job('jobs', $payload);
$factory = $this->getMock('Resque_Job_FactoryInterface');
$testJob = $this->getMock('Resque_JobInterface');
$factory->expects(self::never())->method('create')->will(self::returnValue($testJob));
$instance = $job->getInstance();
$this->assertInstanceOf('Resque_JobInterface', $instance);
}
}
class Some_Job_Class implements Resque_JobInterface
{
/**
* @return bool
*/
public function perform()
{
return true;
}
/**
* @return bool
*/
public function perform()
{
return true;
}
}
class Some_Stub_Factory implements Resque_Job_FactoryInterface
{
/**
* @param $className
* @param array $args
* @param $queue
* @return Resque_JobInterface
*/
public function create($className, $args, $queue)
{
return new Some_Job_Class();
}
/**
* @param $className
* @param array $args
* @param $queue
* @return Resque_JobInterface
*/
public function create($className, $args, $queue)
{
return new Some_Job_Class();
}
}

42
test/Resque/Tests/LogTest.php
View File

@ -1,31 +1,33 @@
<?php
/**
* Resque_Log tests.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_LogTest extends Resque_Tests_TestCase
{
public function testLogInterpolate()
{
$logger = new Resque_Log();
$actual = $logger->interpolate('string {replace}', array('replace' => 'value'));
$expected = 'string value';
public function testLogInterpolate()
{
$logger = new Resque_Log();
$actual = $logger->interpolate('string {replace}', array('replace' => 'value'));
$expected = 'string value';
$this->assertEquals($expected, $actual);
}
$this->assertEquals($expected, $actual);
}
public function testLogInterpolateMutiple()
{
$logger = new Resque_Log();
$actual = $logger->interpolate(
'string {replace1} {replace2}',
array('replace1' => 'value1', 'replace2' => 'value2')
);
$expected = 'string value1 value2';
public function testLogInterpolateMutiple()
{
$logger = new Resque_Log();
$actual = $logger->interpolate(
'string {replace1} {replace2}',
array('replace1' => 'value1', 'replace2' => 'value2')
);
$expected = 'string value1 value2';
$this->assertEquals($expected, $actual);
}
$this->assertEquals($expected, $actual);
}
}

366
test/Resque/Tests/RedisTest.php
View File

@ -1,197 +1,199 @@
<?php
/**
* Resque_Event tests.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_RedisTest extends Resque_Tests_TestCase
{
/**
* @expectedException Resque_RedisException
*/
public function testRedisExceptionsAreSurfaced()
{
$mockCredis = $this->getMockBuilder('Credis_Client')
->setMethods(['connect', '__call'])
->getMock();
$mockCredis->expects($this->any())->method('__call')
->will($this->throwException(new CredisException('failure')));
// public function testRedisExceptionsAreSurfaced()
// {
// $mockCredis = $this->getMockBuilder('Credis_Client')
// ->setMethods(['connect', '__call'])
// ->getMock();
// $mockCredis->expects($this->any())->method('__call')
// ->will($this->throwException(new CredisException('failure')));
//
// Resque::setBackend(function($database) use ($mockCredis) {
// return new Resque_Redis('localhost:6379', $database, $mockCredis);
// });
// Resque::redis()->ping();
// }
Resque::setBackend(function($database) use ($mockCredis) {
return new Resque_Redis('localhost:6379', $database, $mockCredis);
});
Resque::redis()->ping();
}
/**
* These DNS strings are considered valid.
*
* @return array
*/
public function validDsnStringProvider()
{
return [
// Input , Expected output
['', [
'localhost',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
[],
]],
['localhost', [
'localhost',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
[],
]],
['localhost:1234', [
'localhost',
1234,
false,
false, false,
[],
]],
['localhost:1234/2', [
'localhost',
1234,
2,
false, false,
[],
]],
['redis://foobar', [
'foobar',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
[],
]],
['redis://foobar/', [
'foobar',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
[],
]],
['redis://foobar:1234', [
'foobar',
1234,
false,
false, false,
[],
]],
['redis://foobar:1234/15', [
'foobar',
1234,
15,
false, false,
[],
]],
['redis://foobar:1234/0', [
'foobar',
1234,
0,
false, false,
[],
]],
['redis://user@foobar:1234', [
'foobar',
1234,
false,
'user', false,
[],
]],
['redis://user@foobar:1234/15', [
'foobar',
1234,
15,
'user', false,
[],
]],
['redis://user:pass@foobar:1234', [
'foobar',
1234,
false,
'user', 'pass',
[],
]],
['redis://user:pass@foobar:1234?x=y&a=b', [
'foobar',
1234,
false,
'user', 'pass',
['x' => 'y', 'a' => 'b'],
]],
['redis://:pass@foobar:1234?x=y&a=b', [
'foobar',
1234,
false,
false, 'pass',
['x' => 'y', 'a' => 'b'],
]],
['redis://user@foobar:1234?x=y&a=b', [
'foobar',
1234,
false,
'user', false,
['x' => 'y', 'a' => 'b'],
]],
['redis://foobar:1234?x=y&a=b', [
'foobar',
1234,
false,
false, false,
['x' => 'y', 'a' => 'b'],
]],
['redis://user@foobar:1234/12?x=y&a=b', [
'foobar',
1234,
12,
'user', false,
['x' => 'y', 'a' => 'b'],
]],
['tcp://user@foobar:1234/12?x=y&a=b', [
'foobar',
1234,
12,
'user', false,
['x' => 'y', 'a' => 'b'],
]],
];
}
/**
* These DNS strings are considered valid.
*
* @return array
*/
public function validDsnStringProvider()
{
return array(
// Input , Expected output
array('', array(
'localhost',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
array(),
)),
array('localhost', array(
'localhost',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
array(),
)),
array('localhost:1234', array(
'localhost',
1234,
false,
false, false,
array(),
)),
array('localhost:1234/2', array(
'localhost',
1234,
2,
false, false,
array(),
)),
array('redis://foobar', array(
'foobar',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
array(),
)),
array('redis://foobar/', array(
'foobar',
Resque_Redis::DEFAULT_PORT,
false,
false, false,
array(),
)),
array('redis://foobar:1234', array(
'foobar',
1234,
false,
false, false,
array(),
)),
array('redis://foobar:1234/15', array(
'foobar',
1234,
15,
false, false,
array(),
)),
array('redis://foobar:1234/0', array(
'foobar',
1234,
0,
false, false,
array(),
)),
array('redis://user@foobar:1234', array(
'foobar',
1234,
false,
'user', false,
array(),
)),
array('redis://user@foobar:1234/15', array(
'foobar',
1234,
15,
'user', false,
array(),
)),
array('redis://user:pass@foobar:1234', array(
'foobar',
1234,
false,
'user', 'pass',
array(),
)),
array('redis://user:pass@foobar:1234?x=y&a=b', array(
'foobar',
1234,
false,
'user', 'pass',
array('x' => 'y', 'a' => 'b'),
)),
array('redis://:pass@foobar:1234?x=y&a=b', array(
'foobar',
1234,
false,
false, 'pass',
array('x' => 'y', 'a' => 'b'),
)),
array('redis://user@foobar:1234?x=y&a=b', array(
'foobar',
1234,
false,
'user', false,
array('x' => 'y', 'a' => 'b'),
)),
array('redis://foobar:1234?x=y&a=b', array(
'foobar',
1234,
false,
false, false,
array('x' => 'y', 'a' => 'b'),
)),
array('redis://user@foobar:1234/12?x=y&a=b', array(
'foobar',
1234,
12,
'user', false,
array('x' => 'y', 'a' => 'b'),
)),
array('tcp://user@foobar:1234/12?x=y&a=b', array(
'foobar',
1234,
12,
'user', false,
array('x' => 'y', 'a' => 'b'),
)),
);
}
/**
* These DSN values should throw exceptions
* @return array
*/
public function bogusDsnStringProvider()
{
return [
['http://foo.bar/'],
['user:@foobar:1234?x=y&a=b'],
['foobar:1234?x=y&a=b'],
];
}
/**
* These DSN values should throw exceptions
* @return array
*/
public function bogusDsnStringProvider()
{
return array(
array('http://foo.bar/'),
array('user:@foobar:1234?x=y&a=b'),
array('foobar:1234?x=y&a=b'),
);
}
/**
* @dataProvider validDsnStringProvider
*/
public function testParsingValidDsnString($dsn, $expected)
{
$result = Resque_Redis::parseDsn($dsn);
$this->assertEquals($expected, $result);
}
/**
* @dataProvider validDsnStringProvider
*/
public function testParsingValidDsnString($dsn, $expected)
{
$result = Resque_Redis::parseDsn($dsn);
$this->assertEquals($expected, $result);
}
/**
* @dataProvider bogusDsnStringProvider
* @expectedException InvalidArgumentException
*/
public function testParsingBogusDsnStringThrowsException($dsn)
{
// The next line should throw an InvalidArgumentException
$result = Resque_Redis::parseDsn($dsn);
}
/**
* @dataProvider bogusDsnStringProvider
* @expectedException InvalidArgumentException
*/
public function testParsingBogusDsnStringThrowsException($dsn)
{
// The next line should throw an InvalidArgumentException
Resque_Redis::parseDsn($dsn);
}
}

74
test/Resque/Tests/StatTest.php
View File

@ -1,49 +1,51 @@
<?php
/**
* Resque_Stat tests.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_StatTest extends Resque_Tests_TestCase
{
public function testStatCanBeIncremented()
{
Resque_Stat::incr('test_incr');
Resque_Stat::incr('test_incr');
$this->assertEquals(2, $this->redis->get('resque:stat:test_incr'));
}
public function testStatCanBeIncremented()
{
Resque_Stat::incr('test_incr');
Resque_Stat::incr('test_incr');
$this->assertEquals(2, $this->redis->get('resque:stat:test_incr'));
}
public function testStatCanBeIncrementedByX()
{
Resque_Stat::incr('test_incrX', 10);
Resque_Stat::incr('test_incrX', 11);
$this->assertEquals(21, $this->redis->get('resque:stat:test_incrX'));
}
public function testStatCanBeIncrementedByX()
{
Resque_Stat::incr('test_incrX', 10);
Resque_Stat::incr('test_incrX', 11);
$this->assertEquals(21, $this->redis->get('resque:stat:test_incrX'));
}
public function testStatCanBeDecremented()
{
Resque_Stat::incr('test_decr', 22);
Resque_Stat::decr('test_decr');
$this->assertEquals(21, $this->redis->get('resque:stat:test_decr'));
}
public function testStatCanBeDecremented()
{
Resque_Stat::incr('test_decr', 22);
Resque_Stat::decr('test_decr');
$this->assertEquals(21, $this->redis->get('resque:stat:test_decr'));
}
public function testStatCanBeDecrementedByX()
{
Resque_Stat::incr('test_decrX', 22);
Resque_Stat::decr('test_decrX', 11);
$this->assertEquals(11, $this->redis->get('resque:stat:test_decrX'));
}
public function testStatCanBeDecrementedByX()
{
Resque_Stat::incr('test_decrX', 22);
Resque_Stat::decr('test_decrX', 11);
$this->assertEquals(11, $this->redis->get('resque:stat:test_decrX'));
}
public function testGetStatByName()
{
Resque_Stat::incr('test_get', 100);
$this->assertEquals(100, Resque_Stat::get('test_get'));
}
public function testGetStatByName()
{
Resque_Stat::incr('test_get', 100);
$this->assertEquals(100, Resque_Stat::get('test_get'));
}
public function testGetUnknownStatReturns0()
{
$this->assertEquals(0, Resque_Stat::get('test_get_unknown'));
}
public function testGetUnknownStatReturns0()
{
$this->assertEquals(0, Resque_Stat::get('test_get_unknown'));
}
}

42
test/Resque/Tests/TestCase.php
View File

@ -1,30 +1,34 @@
<?php
/**
* Resque test case class. Contains setup and teardown methods.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_TestCase extends PHPUnit_Framework_TestCase
class Resque_Tests_TestCase extends PHPUnit\Framework\TestCase
{
protected $resque;
protected $redis;
protected $resque;
protected $redis;
public static function setUpBeforeClass()
{
date_default_timezone_set('UTC');
}
public static function setUpBeforeClass()
{
date_default_timezone_set('UTC');
}
public function setUp()
{
$config = file_get_contents(REDIS_CONF);
preg_match('#^\s*port\s+([0-9]+)#m', $config, $matches);
$this->redis = new Credis_Client('localhost', $matches[1]);
public function setUp()
{
// $config = file_get_contents(REDIS_CONF);
// preg_match('#^\s*port\s+([0-9]+)#m', $config, $matches);
$this->redis = new \Redis();
$this->redis->connect('localhost');
$this->redis->select(9);
Resque::setBackend('redis://localhost:' . $matches[1]);
Resque::setBackend('localhost', 9);
// Flush redis
$this->redis->flushAll();
}
// Flush redis
$this->redis->flushAll();
}
}

2
test/Resque/Tests/WorkerTest.php
View File

@ -1,4 +1,5 @@
<?php
/**
* Resque_Worker tests.
*
@ -6,6 +7,7 @@
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Tests_WorkerTest extends Resque_Tests_TestCase
{
public function testWorkerRegistersInList()

146
test/bootstrap.php
View File

@ -2,9 +2,9 @@
/**
* Resque test bootstrap file - sets up a test environment.
*
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
* @package Resque/Tests
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
$loader = require __DIR__ . '/../vendor/autoload.php';
@ -15,24 +15,24 @@ define('REDIS_CONF', TEST_MISC . '/redis.conf');
// Attempt to start our own redis instance for tesitng.
exec('which redis-server', $output, $returnVar);
if($returnVar != 0) {
echo "Cannot find redis-server in path. Please make sure redis is installed.\n";
exit(1);
if ($returnVar != 0) {
echo "Cannot find redis-server in path. Please make sure redis is installed.\n";
exit(1);
}
exec('cd ' . TEST_MISC . '; redis-server ' . REDIS_CONF, $output, $returnVar);
usleep(500000);
if($returnVar != 0) {
echo "Cannot start redis-server.\n";
exit(1);
if ($returnVar != 0) {
echo "Cannot start redis-server.\n";
exit(1);
}
// Get redis port from conf
$config = file_get_contents(REDIS_CONF);
if(!preg_match('#^\s*port\s+([0-9]+)#m', $config, $matches)) {
echo "Could not determine redis port from redis.conf";
exit(1);
if (!preg_match('#^\s*port\s+([0-9]+)#m', $config, $matches)) {
echo "Could not determine redis port from redis.conf";
exit(1);
}
Resque::setBackend('localhost:' . $matches[1]);
@ -43,57 +43,59 @@ function killRedis($pid)
if (getmypid() !== $pid) {
return; // don't kill from a forked worker
}
$config = file_get_contents(REDIS_CONF);
if(!preg_match('#^\s*pidfile\s+([^\s]+)#m', $config, $matches)) {
return;
}
$config = file_get_contents(REDIS_CONF);
if (!preg_match('#^\s*pidfile\s+([^\s]+)#m', $config, $matches)) {
return;
}
$pidFile = TEST_MISC . '/' . $matches[1];
if (file_exists($pidFile)) {
$pid = trim(file_get_contents($pidFile));
posix_kill((int) $pid, 9);
$pidFile = TEST_MISC . '/' . $matches[1];
if (file_exists($pidFile)) {
$pid = trim(file_get_contents($pidFile));
posix_kill((int)$pid, 9);
if(is_file($pidFile)) {
unlink($pidFile);
}
}
if (is_file($pidFile)) {
unlink($pidFile);
}
}
// Remove the redis database
if(!preg_match('#^\s*dir\s+([^\s]+)#m', $config, $matches)) {
return;
}
$dir = $matches[1];
// Remove the redis database
if (!preg_match('#^\s*dir\s+([^\s]+)#m', $config, $matches)) {
return;
}
$dir = $matches[1];
if(!preg_match('#^\s*dbfilename\s+([^\s]+)#m', $config, $matches)) {
return;
}
if (!preg_match('#^\s*dbfilename\s+([^\s]+)#m', $config, $matches)) {
return;
}
$filename = TEST_MISC . '/' . $dir . '/' . $matches[1];
if(is_file($filename)) {
unlink($filename);
}
$filename = TEST_MISC . '/' . $dir . '/' . $matches[1];
if (is_file($filename)) {
unlink($filename);
}
}
register_shutdown_function('killRedis', getmypid());
if(function_exists('pcntl_signal')) {
// Override INT and TERM signals, so they do a clean shutdown and also
// clean up redis-server as well.
function sigint()
{
exit;
}
pcntl_signal(SIGINT, 'sigint');
pcntl_signal(SIGTERM, 'sigint');
if (function_exists('pcntl_signal')) {
// Override INT and TERM signals, so they do a clean shutdown and also
// clean up redis-server as well.
function sigint()
{
exit;
}
pcntl_signal(SIGINT, 'sigint');
pcntl_signal(SIGTERM, 'sigint');
}
class Test_Job
{
public static $called = false;
public static $called = false;
public function perform()
{
self::$called = true;
}
public function perform()
{
self::$called = true;
}
}
class Failing_Job_Exception extends Exception
@ -103,10 +105,10 @@ class Failing_Job_Exception extends Exception
class Failing_Job
{
public function perform()
{
throw new Failing_Job_Exception('Message!');
}
public function perform()
{
throw new Failing_Job_Exception('Message!');
}
}
class Test_Job_Without_Perform_Method
@ -116,33 +118,33 @@ class Test_Job_Without_Perform_Method
class Test_Job_With_SetUp
{
public static $called = false;
public $args = false;
public static $called = false;
public $args = false;
public function setUp()
{
self::$called = true;
}
public function setUp()
{
self::$called = true;
}
public function perform()
{
public function perform()
{
}
}
}
class Test_Job_With_TearDown
{
public static $called = false;
public $args = false;
public static $called = false;
public $args = false;
public function perform()
{
public function perform()
{
}
}
public function tearDown()
{
self::$called = true;
}
public function tearDown()
{
self::$called = true;
}
}