idanoo/php-resque
1
0
Fork 0
mirror of https://github.com/idanoo/php-resque synced 2025-07-01 21:52:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

2.0.0 Add namespacing + PHP8.0 support (#1)

Browse source 
2.0.0 (2021-02-19)

Moved to PSR-4
Namespaced codebase
Added more comments throughout
Co-Authored-By: idanoo <daniel@m2.nz>
Co-Committed-By: idanoo <daniel@m2.nz>
This commit is contained in:
idanoo 2021-02-19 12:23:32 +13:00
parent ebec2f7bf7
commit 80d64e79ff
56 changed files with 2215 additions and 1423 deletions
Download patch file Download diff file Expand all files Collapse all files

93
src/Resque/Event.php Normal file
View file

@ -0,0 +1,93 @@
<?php
namespace Resque;
/**
* Resque event/plugin system class
*
* @package Resque/Event
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Event
{
/**
* @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 = [$data];
}
if (empty(self::$events[$event])) {
return true;
}
foreach (self::$events[$event] as $callback) {
if (!is_callable($callback)) {
continue;
}
call_user_func_array($callback, array_values($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 = [];
}
}

16
src/Resque/Exception.php Normal file
View file

@ -0,0 +1,16 @@
<?php
namespace Resque;
/**
* Resque exception.
*
* @package Resque
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Exception extends \Exception
{
}

59
src/Resque/Failure/Failure.php Normal file
View file

@ -0,0 +1,59 @@
<?php
namespace Resque\Failure;
/**
* Failed Resque job.
*
* @package Resque/Failure
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Failure
{
/**
* @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 array $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|string
*/
public static function getBackend()
{
if (self::$backend === null) {
self::$backend = '\\Resque\\Failure\\ResqueFailureRedis';
}
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;
}
}

23
src/Resque/Failure/ResqueFailureInterface.php Normal file
View file

@ -0,0 +1,23 @@
<?php
namespace Resque\Failure;
/**
* Interface that all failure backends should implement.
*
* @package Resque\Failure
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
interface ResqueFailureInterface
{
/**
* 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);
}

37
src/Resque/Failure/ResqueFailureRedis.php Normal file
View file

@ -0,0 +1,37 @@
<?php
namespace Resque\Failure;
/**
* Redis backend for storing failed Resque jobs.
*
* @package Resque\Failure
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class ResqueFailureRedis implements ResqueFailureInterface
{
/**
* 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.
* @throws \Resque\RedisException
*/
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\Resque::redis()->rpush('failed', $data);
}
}

15
src/Resque/Job/DirtyExitException.php Normal file
View file

@ -0,0 +1,15 @@
<?php
namespace Resque\Job;
/**
* Runtime exception class for a job that does not exit cleanly.
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class DirtyExitException extends \RuntimeException
{
}

15
src/Resque/Job/DontCreate.php Normal file
View file

@ -0,0 +1,15 @@
<?php
namespace Resque\Job;
/**
* Exception to be thrown if while enqueuing a job it should not be created.
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class DontCreate extends \Exception
{
}

15
src/Resque/Job/DontPerform.php Normal file
View file

@ -0,0 +1,15 @@
<?php
namespace Resque\Job;
/**
* Exception to be thrown if a job should not be performed/run.
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class DontPerform extends \Exception
{
}

41
src/Resque/Job/Factory.php Normal file
View file

@ -0,0 +1,41 @@
<?php
namespace Resque\Job;
/**
* Job Factory!
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Factory implements FactoryInterface
{
/**
* @param $className
* @param array $args
* @param $queue
* @return \Resque\Job\JobInterface
* @throws \Resque\Exception
*/
public function create($className, $args, $queue)
{
if (!class_exists($className)) {
throw new \Resque\Exception(
'Could not find job class ' . $className . '.'
);
}
if (!method_exists($className, 'perform')) {
throw new \Resque\Exception(
'Job class ' . $className . ' does not contain a perform method.'
);
}
$instance = new $className();
$instance->args = $args;
$instance->queue = $queue;
return $instance;
}
}

21
src/Resque/Job/FactoryInterface.php Normal file
View file

@ -0,0 +1,21 @@
<?php
namespace Resque\Job;
/**
* Job Interface
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
interface FactoryInterface
{
/**
* @param $className
* @param array $args
* @param $queue
* @return \Resque\Job\JobInterface
*/
public function create($className, $args, $queue);
}

291
src/Resque/Job/Job.php Executable file
View file

@ -0,0 +1,291 @@
<?php
namespace Resque\Job;
/**
* Resque job.
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Job
{
/**
* @var string The name of the queue that this job belongs to.
*/
public $queue;
/**
* @var \Resque\Worker Instance of the Resque worker running this job.
*/
public $worker;
/**
* @var array Array containing details of the job.
*/
public $payload;
/**
* @var object|\Resque\Job\JobInterface Instance of the class performing work for this job.
*/
private $instance;
/**
* @var \Resque\Job\FactoryInterface
*/
private $jobFactory;
/**
* Instantiate a new instance of a job.
*
* @param string $queue The queue that the job belongs to.
* @param array $payload array containing details of the job.
*/
public function __construct($queue, $payload)
{
$this->queue = $queue;
$this->payload = $payload;
}
/**
* 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 $monitor Set to true to be able to monitor the status of a job.
* @param string $id Unique identifier for tracking the job. Generated if not supplied.
*
* @return string
* @throws \InvalidArgumentException
*/
public static function create($queue, $class, $args = null, $monitor = false, $id = null)
{
if (is_null($id)) {
$id = \Resque\Resque::generateJobId();
}
if ($args !== null && !is_array($args)) {
throw new \InvalidArgumentException(
'Supplied $args must be an array.'
);
}
\Resque\Resque::push($queue, [
'class' => $class,
'args' => [$args],
'id' => $id,
'queue_time' => microtime(true),
]);
if ($monitor) {
Status::create($id);
}
return $id;
}
/**
* Find the next available job from the specified queue and return an
* instance of \Resque\Job\Job for it.
*
* @param string $queue The name of the queue to check for a job in.
* @return false|object Null when there aren't any waiting jobs, instance of \Resque\Job\Job when a job was found.
*/
public static function reserve($queue)
{
$payload = \Resque\Resque::pop($queue);
if (!is_array($payload)) {
return false;
}
return new Job($queue, $payload);
}
/**
* Find the next available job from the specified queues using blocking list pop
* and return an instance of \Resque\Job\Job for it.
*
* @param array $queues
* @param int $timeout
* @return false|object Null when there aren't any waiting jobs, instance of \Resque\Job\Job when a job was found.
*/
public static function reserveBlocking(array $queues, $timeout = null)
{
$item = \Resque\Resque::blpop($queues, $timeout);
if (!is_array($item)) {
return false;
}
return new Job($item['queue'], $item['payload']);
}
/**
* Update the status of the current job.
*
* @param int $status Status constant from \Resque\Job\Status indicating the current status of a job.
*/
public function updateStatus($status): bool
{
if (empty($this->payload['id'])) {
return false;
}
$statusInstance = new Status($this->payload['id']);
$statusInstance->update($status);
return true;
}
/**
* Return the status of the current job.
*
* @return int The status of the job as one of the \Resque\Job\Status constants.
*/
public function getStatus()
{
$status = new Status($this->payload['id']);
return $status->get();
}
/**
* Get the arguments supplied to this job.
*
* @return array Array of arguments.
*/
public function getArguments()
{
if (!isset($this->payload['args'])) {
return [];
}
return $this->payload['args'][0];
}
/**
* Get the instantiated object for this job that will be performing work.
* @return \Resque\Job\JobInterface Instance of the object that this job belongs to.
*/
public function getInstance()
{
if (!is_null($this->instance)) {
return $this->instance;
}
$this->instance = $this->getJobFactory()->create($this->payload['class'], $this->getArguments(), $this->queue);
$this->instance->job = $this;
return $this->instance;
}
/**
* Actually execute a job by calling the perform method on the class
* associated with the job with the supplied arguments.
*
* @return bool
* @throws \Resque\Exception When the job's class could not be found or it does not contain a perform method.
*/
public function perform()
{
try {
\Resque\Event::trigger('beforePerform', $this);
$instance = $this->getInstance();
if (method_exists($instance, 'setUp')) {
$instance->setUp();
}
$instance->perform();
if (method_exists($instance, 'tearDown')) {
$instance->tearDown();
}
\Resque\Event::trigger('afterPerform', $this);
} catch (DontPerform $e) {
/** @noinspection PhpRedundantCatchClauseInspection */
return false;
}
return true;
}
/**
* Mark the current job as having failed.
*
* @param $exception
*/
public function fail($exception)
{
\Resque\Event::trigger('onFailure', [
'exception' => $exception,
'job' => $this,
]);
$this->updateStatus(Status::STATUS_FAILED);
\Resque\Failure\Failure::create(
$this->payload,
$exception,
$this->worker,
$this->queue
);
\Resque\Stat::incr('failed');
\Resque\Stat::incr('failed:' . $this->worker);
}
/**
* Re-queue the current job.
* @return string
*/
public function recreate()
{
$status = new Status($this->payload['id']);
$monitor = false;
if ($status->isTracking()) {
$monitor = true;
}
return self::create($this->queue, $this->payload['class'], $this->getArguments(), $monitor);
}
/**
* Generate a string representation used to describe the current job.
*
* @return string The string representation of the job.
*/
public function __toString()
{
$name = [
'Job{' . $this->queue . '}'
];
if (!empty($this->payload['id'])) {
$name[] = 'ID: ' . $this->payload['id'];
}
$name[] = $this->payload['class'];
if (!empty($this->payload['args'])) {
$name[] = json_encode($this->payload['args']);
}
return '(' . implode(' | ', $name) . ')';
}
/**
* @param FactoryInterface $jobFactory
* @return Job
*/
public function setJobFactory(FactoryInterface $jobFactory)
{
$this->jobFactory = $jobFactory;
return $this;
}
/**
* @return FactoryInterface
*/
public function getJobFactory()
{
if ($this->jobFactory === null) {
$this->jobFactory = new Factory();
}
return $this->jobFactory;
}
}

11
src/Resque/Job/JobInterface.php Normal file
View file

@ -0,0 +1,11 @@
<?php
namespace Resque\Job;
interface JobInterface
{
/**
* @return bool
*/
public function perform();
}

146
src/Resque/Job/Status.php Normal file
View file

@ -0,0 +1,146 @@
<?php
namespace Resque\Job;
/**
* Status tracker/information for a job.
*
* @package Resque/Job
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Status
{
public const STATUS_WAITING = 1;
public const STATUS_RUNNING = 2;
public const STATUS_FAILED = 3;
public const STATUS_COMPLETE = 4;
/**
* @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 array Array of statuses that are considered final/complete.
*/
private static $completeStatuses = [
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;
}
/**
* 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 = [
'status' => self::STATUS_WAITING,
'updated' => time(),
'started' => time(),
];
\Resque\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(): bool
{
if ($this->isTracking === false) {
return false;
}
if (!\Resque\Resque::redis()->exists((string)$this)) {
$this->isTracking = false;
return false;
}
$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;
}
$statusPacket = [
'status' => $status,
'updated' => time(),
];
\Resque\Resque::redis()->set((string)$this, json_encode($statusPacket));
// Expire the status for completed jobs after 24 hours
if (in_array($status, self::$completeStatuses)) {
\Resque\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;
}
$statusPacket = json_decode(\Resque\Resque::redis()->get((string)$this), true);
if (!$statusPacket) {
return false;
}
return $statusPacket['status'];
}
/**
* Stop tracking the status of a job.
*/
public function stop()
{
\Resque\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(): string
{
return 'job:' . $this->id . ':status';
}
}

76
src/Resque/Log.php Normal file
View file

@ -0,0 +1,76 @@
<?php
namespace Resque;
/**
* Resque default logger PSR-3 compliant
*
* @package Resque
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Log extends \Psr\Log\AbstractLogger
{
public $logLevel;
public function __construct($logLevel = "warning")
{
$this->logLevel = strtolower($logLevel);
}
/**
* 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 = [])
{
$logLevels = [
"emergency",
"alert",
"critical",
"error",
"warning",
"notice",
"info",
"debug",
];
/**
* Only log things with a higher level than the current log level.
* e.g If set as 'alert' will only alert for 'emergency' and 'alert' logs.
*/
if (array_search($level, $logLevels) <= array_search($this->logLevel, $logLevels)) {
fwrite(
STDOUT,
'[' . $level . '][' . strftime('%Y-%m-%d %T') . '] ' .
$this->interpolate($message, $context) . PHP_EOL
);
}
return;
}
/**
* 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 = [])
{
// build a replacement array with braces around the context keys
$replace = [];
foreach ($context as $key => $val) {
$replace['{' . $key . '}'] = $val;
}
// interpolate replacement values into the message and return
return strtr($message, $replace);
}
}

271
src/Resque/Redis.php Normal file
View file

@ -0,0 +1,271 @@
<?php
namespace Resque;
/**
* Set up phpredis connection
*
* @package Resque/Redis
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Redis
{
/**
* Redis Client
* @var \Credis_Client
*/
private $driver;
/**
* Redis namespace
* @var string
*/
private static $defaultNamespace = 'resque:';
/**
* A default host to connect to
*/
public const DEFAULT_HOST = 'localhost';
/**
* The default Redis port
*/
public const DEFAULT_PORT = 6379;
/**
* The default Redis Database number
*/
public 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 = [
'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;
}
/**
* @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_Client instance instantiated by you
* @throws \Resque\RedisException
*/
public function __construct($server, $database = null, $client = null)
{
try {
if (is_object($client)) {
$this->driver = $client;
} else {
/** @noinspection PhpUnusedLocalVariableInspection */
list($host, $port, $dsnDatabase, $user, $password, $options) = self::parseDsn($server);
// $user is not used, only $password
$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;
$this->driver = new \Credis_Client($host, $port, $timeout, $persistent);
$this->driver->setMaxConnectRetries($maxRetries);
if ($password) {
$this->driver->auth($password);
}
// 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 ($database !== null) {
$this->driver->select($database);
}
} catch (\Exception $e) {
throw new RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
}
}
/**
* 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);
// Check the URI scheme
$validSchemes = ['redis', 'tcp'];
if (isset($parts['scheme']) && !in_array($parts['scheme'], $validSchemes)) {
throw new \InvalidArgumentException("Invalid DSN. Supported schemes are " . implode(', ', $validSchemes));
}
// Allow simple 'hostname' format, which `parse_url` treats as a path, not host.
if (!isset($parts['host']) && isset($parts['path'])) {
$parts['host'] = $parts['path'];
unset($parts['path']);
}
// Extract the port number as an integer
$port = isset($parts['port']) ? intval($parts['port']) : self::DEFAULT_PORT;
// 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']));
}
// Extract any 'user' and 'pass' values
$user = isset($parts['user']) ? $parts['user'] : false;
$pass = isset($parts['pass']) ? $parts['pass'] : false;
// Convert the query string into an associative array
$options = [];
if (isset($parts['query'])) {
// Parse the query string into an array
parse_str($parts['query'], $options);
}
return [
$parts['host'],
$port,
$database,
$user,
$pass,
$options,
];
}
/**
* 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.
* @throws Resque_RedisException
*/
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 RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
}
}
public static function getPrefix(): string
{
return self::$defaultNamespace;
}
public static function removePrefix($string): string
{
$prefix = self::getPrefix();
if (substr($string, 0, strlen($prefix)) == $prefix) {
$string = substr($string, strlen($prefix), strlen($string));
}
return $string;
}
}

16
src/Resque/RedisException.php Normal file
View file

@ -0,0 +1,16 @@
<?php
namespace Resque;
/**
* Redis related exceptions
*
* @package Resque
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class RedisException extends \Exception
{
}

395
src/Resque/Resque.php Normal file
View file

@ -0,0 +1,395 @@
<?php
namespace Resque;
/**
* Base Resque class.
*
* @package Resque
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque
{
public const VERSION = '2.0.0';
public const DEFAULT_INTERVAL = 5;
/**
* @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 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;
}
/**
* Return an instance of the Resque_Redis class instantiated for Resque.
*
* @return \Resque\Redis Instance of Resque_Redis.
*
* @throws \Resque\RedisException
*/
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);
}
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;
}
// 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.');
}
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.
*
* @return bool
*/
public static function push($queue, $item): bool
{
$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 mixed Decoded item from the queue.
*/
public static function pop($queue)
{
$item = self::redis()->lpop('queue:' . $queue);
if (!$item) {
return false;
}
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 = [])
{
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 int Number of deleted items
*/
public static function removeQueue($queue): int
{
$num = self::removeList($queue);
self::redis()->srem('queues', $queue);
return intval($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 array|null|void
*/
public static function blpop(array $queues, $timeout)
{
$list = [];
foreach ($queues as $queue) {
$list[] = 'queue:' . $queue;
}
$item = self::redis()->blpop($list, (int)$timeout);
if (!$item) {
return false;
}
/**
* 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 [
'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): int
{
return intval(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 = [
'class' => $class,
'args' => $args,
'queue' => $queue,
'id' => $id,
];
try {
\Resque\Event::trigger('beforeEnqueue', $hookParams);
} catch (\Resque\Job\DontCreate $e) {
return false;
}
\Resque\Job\Job::create($queue, $class, $args, $trackStatus, $id);
\Resque\Event::trigger('afterEnqueue', $hookParams);
return $id;
}
/**
* Reserve and return the next available job in the specified queue.
*
* @param string $queue Queue to fetch next available job from.
*
* @return false|object|\Resque\Job\Job
*/
public static function reserve($queue)
{
return \Resque\Job\Job::reserve($queue);
}
/**
* Get an array of all known queues.
*
* @return array Array of queues.
*/
public static function queues(): array
{
$queues = self::redis()->smembers('queues');
return is_array($queues) ? $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 int number of deleted items
*/
private static function removeItems($queue, $items = []): int
{
$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);
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;
}
}
// remove temp queue and requeue queue
self::redis()->del($requeueQueue);
self::redis()->del($tempQueue);
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
*
* @param $string
* @param $items
* @return bool (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;
}
/**
* Remove List
*
* @private
*
* @params string $queue the name of the queue
* @param $queue
* @return integer number of deleted items belongs to this list
* @throws \Resque\RedisException
*/
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));
}
}

60
src/Resque/Stat.php Normal file
View file

@ -0,0 +1,60 @@
<?php
namespace Resque;
/**
* Resque statistic management (jobs processed, failed, etc)
*
* @package Resque
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class 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): int
{
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): bool
{
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): bool
{
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): bool
{
return (bool)Resque::redis()->del('stat:' . $stat);
}
}

583
src/Resque/Worker.php Normal file
View file

@ -0,0 +1,583 @@
<?php
declare(ticks=1);
namespace Resque;
use Psr\Log\LoggerInterface;
/**
* Resque worker that handles checking queues for jobs, fetching them
* off the queues, running them and handling the result.
*
* @package Resque
* @author Daniel Mason <daniel@m2.nz>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Worker
{
/**
* @var LoggerInterface Logging object that impliments the PSR-3 LoggerInterface
*/
public $logger;
/**
* @var array Array of all associated queues for this worker.
*/
private $queues = [];
/**
* @var string The hostname of this worker.
*/
private $hostname;
/**
* @var boolean True if on the next iteration, the worker should shutdown.
*/
private $shutdown = false;
/**
* @var boolean True if this worker is paused.
*/
private $paused = false;
/**
* @var string String identifying this worker.
*/
private $id;
/**
* @var \Resque\Job\Job Current job, if any, being processed by this worker.
*/
private $currentJob = null;
/**
* @var int Process ID of child worker processes.
*/
private $child = null;
/**
* Instantiate a new worker, given a list of queues that it should be working
* on. The list of queues should be supplied in the priority that they should
* be checked for jobs (first come, first served)
*
* Passing a single '*' allows the worker to work on all queues in alphabetical
* order. You can easily add new queues dynamically and have them worked on using
* this method.
*
* @param string|array $queues String with a single queue name, array with multiple.
*/
public function __construct($queues)
{
$this->logger = new Log();
if (!is_array($queues)) {
$queues = [$queues];
}
$this->queues = $queues;
$this->hostname = php_uname('n');
$this->id = $this->hostname . ':' . getmypid() . ':' . implode(',', $this->queues);
}
/**
* Return all workers known to Resque as instantiated instances.
* @return array
*/
public static function all(): array
{
$workers = Resque::redis()->smembers('workers');
if (!is_array($workers)) {
$workers = [];
}
$instances = [];
foreach ($workers as $workerId) {
$instances[] = self::find($workerId);
}
return $instances;
}
/**
* Given a worker ID, check if it is registered/valid.
*
* @param string $workerId ID of the worker.
* @return boolean True if the worker exists, false if not.
* @throws Resque_RedisException
*/
public static function exists($workerId): bool
{
return (bool)Resque::redis()->sismember('workers', $workerId);
}
/**
* Given a worker ID, find it and return an instantiated worker class for it.
*
* @param string $workerId The ID of the worker.
* @return bool|Resque_Worker
* @throws Resque_RedisException
*/
public static function find($workerId)
{
if (false === strpos($workerId, ":") || !self::exists($workerId)) {
return false;
}
/** @noinspection PhpUnusedLocalVariableInspection */
list($hostname, $pid, $queues) = explode(':', $workerId, 3);
$queues = explode(',', $queues);
$worker = new self($queues);
$worker->setId($workerId);
return $worker;
}
/**
* Set the ID of this worker to a given ID string.
*
* @param string $workerId ID for the worker.
*/
public function setId($workerId)
{
$this->id = $workerId;
}
/**
* The primary loop for a worker which when called on an instance starts
* the worker's life cycle.
*
* Queues are checked every $interval (seconds) for new jobs.
*
* @param int $interval How often to check for new jobs across the queues.
* @param bool $blocking
* @throws Resque_RedisException
*/
public function work($interval = Resque::DEFAULT_INTERVAL, $blocking = false)
{
$this->updateProcLine('Starting');
$this->startup();
while (true) {
if ($this->shutdown) {
break;
}
// Attempt to find and reserve a job
$job = false;
if (!$this->paused) {
if ($blocking === true) {
$this->logger->log(
\Psr\Log\LogLevel::INFO,
'Starting blocking with timeout of {interval}',
['interval' => $interval],
);
$this->updateProcLine(
'Waiting for ' . implode(',', $this->queues) . ' with blocking timeout ' . $interval
);
} else {
$this->updateProcLine('Waiting for ' . implode(',', $this->queues) . ' with interval ' . $interval);
}
$job = $this->reserve($blocking, $interval);
}
if (!$job) {
// For an interval of 0, break now - helps with unit testing etc
if ($interval == 0) {
break;
}
if ($blocking === false) {
// If no job was found, we sleep for $interval before continuing and checking again
$this->logger->log(
\Psr\Log\LogLevel::INFO,
'Sleeping for {interval}',
['interval' => $interval],
);
if ($this->paused) {
$this->updateProcLine('Paused');
} else {
$this->updateProcLine('Waiting for ' . implode(',', $this->queues));
}
usleep($interval * 1000000);
}
continue;
}
$this->logger->log(\Psr\Log\LogLevel::NOTICE, 'Starting work on {job}', ['job' => $job]);
Event::trigger('beforeFork', $job);
$this->workingOn($job);
$this->child = Resque::fork();
// Forked and we're the child. Run the job.
if ($this->child === 0 || $this->child === false) {
$status = 'Processing ' . $job->queue . ' since ' . strftime('%F %T');
$this->updateProcLine($status);
$this->logger->log(\Psr\Log\LogLevel::INFO, $status);
/** @noinspection PhpParamsInspection */
$this->perform($job);
if ($this->child === 0) {
exit(0);
}
}
if ($this->child > 0) {
// Parent process, sit and wait
$status = 'Forked ' . $this->child . ' at ' . strftime('%F %T');
$this->updateProcLine($status);
$this->logger->log(\Psr\Log\LogLevel::INFO, $status);
// Wait until the child process finishes before continuing
pcntl_wait($status);
$exitStatus = pcntl_wexitstatus($status);
if ($exitStatus !== 0) {
$job->fail(new \Resque\Job\DirtyExitException(
'Job exited with exit code ' . $exitStatus
));
}
}
$this->child = null;
$this->doneWorking();
}
$this->unregisterWorker();
}
/**
* Process a single job.
*
* @param \Resque\Job\Job $job The job to be processed.
*/
public function perform(\Resque\Job\Job $job)
{
try {
Event::trigger('afterFork', $job);
$job->perform();
} catch (\Exception $e) {
$this->logger->log(\Psr\Log\LogLevel::CRITICAL, '{job} has failed {stack}', ['job' => $job, 'stack' => $e]);
$job->fail($e);
return;
}
$job->updateStatus(\Resque\Job\Status::STATUS_COMPLETE);
$this->logger->log(\Psr\Log\LogLevel::NOTICE, '{job} has finished', ['job' => $job]);
}
/**
* @param bool $blocking
* @param int $timeout
* @return object|boolean Instance of \Resque\Job\Job if a job is found, false if not.
*/
public function reserve($blocking = false, $timeout = null)
{
$queues = $this->queues();
if (!is_array($queues)) {
return false;
}
if ($blocking === true) {
$job = \Resque\Job\Job::reserveBlocking($queues, $timeout);
if ($job) {
$this->logger->log(\Psr\Log\LogLevel::INFO, 'Found job on {queue}', ['queue' => $job->queue]);
return $job;
}
} else {
foreach ($queues as $queue) {
$this->logger->log(\Psr\Log\LogLevel::INFO, 'Checking {queue} for jobs', ['queue' => $queue]);
$job = \Resque\Job\Job::reserve($queue);
if ($job) {
$this->logger->log(\Psr\Log\LogLevel::INFO, 'Found job on {queue}', ['queue' => $job->queue]);
return $job;
}
}
}
return false;
}
/**
* Return an array containing all of the queues that this worker should use
* when searching for jobs.
*
* If * is found in the list of queues, every queue will be searched in
* alphabetic order. (@param boolean $fetch If true, and the queue is set to *, will fetch
* all queue names from redis.
* @return array Array of associated queues.
* @see $fetch)
*
*/
public function queues($fetch = true)
{
if (!in_array('*', $this->queues) || $fetch == false) {
return $this->queues;
}
$queues = Resque::queues();
sort($queues);
return $queues;
}
/**
* Perform necessary actions to start a worker.
*/
private function startup()
{
$this->registerSigHandlers();
$this->pruneDeadWorkers();
Event::trigger('beforeFirstFork', $this);
$this->registerWorker();
}
/**
* On supported systems (with the PECL proctitle module installed), update
* the name of the currently running process to indicate the current state
* of a worker.
*
* @param string $status The updated process title.
*/
private function updateProcLine($status)
{
$processTitle = 'resque-' . Resque::VERSION . ': ' . $status;
if (function_exists('cli_set_process_title') && PHP_OS !== 'Darwin') {
cli_set_process_title($processTitle);
} elseif (function_exists('setproctitle')) {
setproctitle($processTitle);
}
}
/**
* Register signal handlers that a worker should respond to.
*
* TERM: Shutdown immediately and stop processing jobs.
* INT: Shutdown immediately and stop processing jobs.
* QUIT: Shutdown after the current job finishes processing.
* USR1: Kill the forked child immediately and continue processing jobs.
*/
private function registerSigHandlers()
{
if (!function_exists('pcntl_signal')) {
return;
}
pcntl_signal(SIGTERM, [$this, 'shutDownNow']);
pcntl_signal(SIGINT, [$this, 'shutDownNow']);
pcntl_signal(SIGQUIT, [$this, 'shutdown']);
pcntl_signal(SIGUSR1, [$this, 'killChild']);
pcntl_signal(SIGUSR2, [$this, 'pauseProcessing']);
pcntl_signal(SIGCONT, [$this, 'unPauseProcessing']);
$this->logger->log(\Psr\Log\LogLevel::DEBUG, 'Registered signals');
}
/**
* Signal handler callback for USR2, pauses processing of new jobs.
*/
public function pauseProcessing()
{
$this->logger->log(\Psr\Log\LogLevel::NOTICE, 'USR2 received; pausing job processing');
$this->paused = true;
}
/**
* Signal handler callback for CONT, resumes worker allowing it to pick
* up new jobs.
*/
public function unPauseProcessing()
{
$this->logger->log(\Psr\Log\LogLevel::NOTICE, 'CONT received; resuming job processing');
$this->paused = false;
}
/**
* Schedule a worker for shutdown. Will finish processing the current job
* and when the timeout interval is reached, the worker will shut down.
*/
public function shutdown()
{
$this->shutdown = true;
$this->logger->log(\Psr\Log\LogLevel::NOTICE, 'Shutting down');
}
/**
* Force an immediate shutdown of the worker, killing any child jobs
* currently running.
*/
public function shutdownNow()
{
$this->shutdown();
$this->killChild();
}
/**
* Kill a forked child job immediately. The job it is processing will not
* be completed.
*/
public function killChild()
{
if (!$this->child) {
$this->logger->log(\Psr\Log\LogLevel::DEBUG, 'No child to kill.');
return;
}
$this->logger->log(\Psr\Log\LogLevel::INFO, 'Killing child at {child}', ['child' => $this->child]);
if (exec('ps -o pid,state -p ' . $this->child, $output, $returnCode) && $returnCode != 1) {
$this->logger->log(\Psr\Log\LogLevel::DEBUG, 'Child {child} found, killing.', ['child' => $this->child]);
posix_kill($this->child, SIGKILL);
$this->child = null;
} else {
$this->logger->log(
\Psr\Log\LogLevel::INFO,
'Child {child} not found, restarting.',
['child' => $this->child],
);
$this->shutdown();
}
}
/**
* Look for any workers which should be running on this server and if
* they're not, remove them from Redis.
*
* This is a form of garbage collection to handle cases where the
* server may have been killed and the Resque workers did not die gracefully
* and therefore leave state information in Redis.
*/
public function pruneDeadWorkers()
{
$workerPids = $this->workerPids();
$workers = self::all();
foreach ($workers as $worker) {
if (is_object($worker)) {
list($host, $pid, $queues) = explode(':', (string)$worker, 3);
if ($host != $this->hostname || in_array($pid, $workerPids) || $pid == getmypid()) {
continue;
}
$this->logger->log(
\Psr\Log\LogLevel::INFO,
'Pruning dead worker: {worker}',
['worker' => (string)$worker],
);
$worker->unregisterWorker();
}
}
}
/**
* Return an array of process IDs for all of the Resque workers currently
* running on this machine.
*
* @return array Array of Resque worker process IDs.
*/
public function workerPids()
{
$pids = [];
exec('ps -A -o pid,command | grep [r]esque', $cmdOutput);
foreach ($cmdOutput as $line) {
list($pids[],) = explode(' ', trim($line), 2);
}
return $pids;
}
/**
* Register this worker in Redis.
* 48 hour TTL so we don't pollute the db on server termination.
*/
public function registerWorker()
{
Resque::redis()->sadd('workers', (string)$this);
Resque::redis()->setex('worker:' . (string)$this . ':started', 172800, strftime('%a %b %d %H:%M:%S %Z %Y'));
}
/**
* Unregister this worker in Redis. (shutdown etc)
*/
public function unregisterWorker()
{
if (is_object($this->currentJob)) {
$this->currentJob->fail(new \Resque\Job\DirtyExitException());
}
$id = (string)$this;
Resque::redis()->srem('workers', $id);
Resque::redis()->del('worker:' . $id);
Resque::redis()->del('worker:' . $id . ':started');
Stat::clear('processed:' . $id);
Stat::clear('failed:' . $id);
}
/**
* Tell Redis which job we're currently working on.
*
* @param \Resque\Job\Job $job \Resque\Job\Job instance containing the job we're working on.
* @throws Resque_RedisException
*/
public function workingOn(\Resque\Job\Job $job)
{
$job->worker = $this;
$this->currentJob = $job;
$job->updateStatus(\Resque\Job\Status::STATUS_RUNNING);
$data = json_encode([
'queue' => $job->queue,
'run_at' => strftime('%a %b %d %H:%M:%S %Z %Y'),
'payload' => $job->payload
]);
Resque::redis()->set('worker:' . $job->worker, $data);
}
/**
* Notify Redis that we've finished working on a job, clearing the working
* state and incrementing the job stats.
*/
public function doneWorking()
{
$this->currentJob = null;
Stat::incr('processed');
Stat::incr('processed:' . (string)$this);
Resque::redis()->del('worker:' . (string)$this);
}
/**
* Generate a string representation of this worker.
*
* @return string String identifier for this worker instance.
*/
public function __toString()
{
return $this->id;
}
/**
* Return an object describing the job this worker is currently working on.
*
* @return array Array with details of current job.
*/
public function job(): array
{
$job = Resque::redis()->get('worker:' . $this);
return $job ? json_decode($job, true) : [];
}
/**
* Get a statistic belonging to this worker.
*
* @param string $stat Statistic to fetch.
* @return int Statistic value.
*/
public function getStat($stat)
{
return \Resque\Stat::get($stat . ':' . $this);
}
/**
* Inject the logging object into the worker
*
* @param \Psr\Log\LoggerInterface $logger
*/
public function setLogger(\Psr\Log\LoggerInterface $logger)
{
$this->logger = $logger;
}
}