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

- Updated travis builds to run on PHP 7.0, 7.1 and 7.2.

Browse Source 
- Added ability to specify multiple log levels. [DEBUG/INFO/NOTICE/WARNING/ERROR/CRITICAL/ALERT/EMERGENCY]
- Default  is now .
- Removed VERBOSE / VVERBOSE flags.
- Enabled date/time logging by default.
This commit is contained in:
Daniel Mason 2018-05-29 22:20:56 +12:00
parent 177f8e3c19
commit 0d0c0d0a7e
8 changed files with 68 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.travis.yml
View File

@ -4,7 +4,6 @@ php:
- 7.0
- 7.1
- 7.2
- hhvm
services:
- redis-server

11
CHANGELOG.md
View File

@ -1,6 +1,13 @@
## 1.4.1 (2018-05-29)
- Updated travis builds to run on PHP 7.0, 7.1 and 7.2.
- Added ability to specify multiple log levels. [DEBUG/INFO/NOTICE/WARNING/ERROR/CRITICAL/ALERT/EMERGENCY]
- Default `LOGLEVEL` is now `WARNING`.
- Removed VERBOSE / VVERBOSE flags.
- Enabled date/time logging by default.
## 1.4 (2018-05-25)
- Replaced credis (rather unmaintained) in favour of phpredis.
- Reformatted codebase to be PSR2 compliant.
## 1.3 (2013) ##

13
README.md
View File

@ -36,7 +36,7 @@ pre and post jobs
## Requirements ##
* PHP 5.6+
* PHP 7.0+ (May work with 5.6+, Untested)
* phpredis
* Redis 2.2+
@ -232,12 +232,15 @@ variables are set (`setenv`) before you do.
### Logging ###
The port supports the same environment variables for logging to STDOUT.
Setting `VERBOSE` will print basic debugging information and `VVERBOSE`
will print detailed information.
Setting `LOGLEVEL` will print different logs depending on levels.
Valid loglevels are listed below with an example.
Default `LOGLEVEL` is `WARNING`.
```bash
[DEBUG/INFO/NOTICE/WARNING/ERROR/CRITICAL/ALERT/EMERGENCY]
```
```sh
$ VERBOSE=1 QUEUE=file_serve bin/resque
$ VVERBOSE=1 QUEUE=file_serve bin/resque
$ LOGLEVEL=DEBUG QUEUE=file_serve bin/resque
```
### Priorities and Queue Lists ###

10
bin/resque
View File

@ -48,13 +48,9 @@ if (!empty($REDIS_BACKEND)) {
}
$logLevel = false;
$LOGGING = getenv('LOGGING');
$VERBOSE = getenv('VERBOSE');
$VVERBOSE = getenv('VVERBOSE');
if (!empty($LOGGING) || !empty($VERBOSE)) {
$logLevel = true;
} else if (!empty($VVERBOSE)) {
$logLevel = true;
$LOGGING = getenv('LOGLEVEL');
if (!empty($LOGGING) ) {
$logLevel = $LOGGING;
}
$APP_INCLUDE = getenv('APP_INCLUDE');

8
demo/queue.php
View File

@ -11,12 +11,12 @@ Resque::setBackend('127.0.0.1:6379');
//Resque::setBackend('redis://user:pass@127.0.0.1:6379');
//Resque::setBackend('redis://user:pass@a.host.name:3432/2');
$args = array(
$args = [
'time' => time(),
'array' => array(
'array' => [
'test' => 'test',
),
);
],
];
if (empty($argv[2])) {
$jobId = Resque::enqueue('default', $argv[1], $args, true);
} else {

25
lib/Resque/Log.php
View File

@ -10,11 +10,11 @@
class Resque_Log extends Psr\Log\AbstractLogger
{
public $verbose;
public $logLevel;
public function __construct($verbose = false)
public function __construct($logLevel = "warning")
{
$this->verbose = $verbose;
$this->logLevel = strtolower($logLevel);
}
/**
@ -25,22 +25,17 @@ class Resque_Log extends Psr\Log\AbstractLogger
* @param array $context Variables to replace { placeholder }
* @return null
*/
public function log($level, $message, array $context = array())
public function log($level, $message, array $context = [])
{
if ($this->verbose) {
$logLevels = ["emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"];
if (array_search($level, $logLevels) <= array_search($this->logLevel, $logLevels)) {
fwrite(
STDOUT,
'[' . $level . '] [' . strftime('%T %Y-%m-%d') . '] ' . $this->interpolate($message, $context) . PHP_EOL
'[' . $level . '][' . strftime('%T %Y-%m-%d') . '] ' . $this->interpolate($message, $context) . PHP_EOL
);
return;
}
return;
if (!($level === Psr\Log\LogLevel::INFO || $level === Psr\Log\LogLevel::DEBUG)) {
fwrite(
STDOUT,
'[' . $level . '] ' . $this->interpolate($message, $context) . PHP_EOL
);
}
}
/**
@ -51,10 +46,10 @@ class Resque_Log extends Psr\Log\AbstractLogger
* @param array $context Array of variables to use in message
* @return string
*/
public function interpolate($message, array $context = array())
public function interpolate($message, array $context = [])
{
// build a replacement array with braces around the context keys
$replace = array();
$replace = [];
foreach ($context as $key => $val) {
$replace['{' . $key . '}'] = $val;
}

10
lib/Resque/Redis.php
View File

@ -181,7 +181,7 @@ class Resque_Redis
$parts = parse_url($dsn);
// Check the URI scheme
$validSchemes = array('redis', 'tcp');
$validSchemes = ['redis', 'tcp'];
if (isset($parts['scheme']) && !in_array($parts['scheme'], $validSchemes)) {
throw new \InvalidArgumentException("Invalid DSN. Supported schemes are " . implode(', ', $validSchemes));
}
@ -207,20 +207,20 @@ class Resque_Redis
$pass = isset($parts['pass']) ? $parts['pass'] : false;
// Convert the query string into an associative array
$options = array();
$options = [];
if (isset($parts['query'])) {
// Parse the query string into an array
parse_str($parts['query'], $options);
}
return array(
return [
$parts['host'],
$port,
$database,
$user,
$pass,
$options,
);
];
}
/**
@ -244,7 +244,7 @@ class Resque_Redis
}
}
try {
return call_user_func_array(array($this->redisConnection, $name), $args);
return call_user_func_array([$this->redisConnection, $name], $args);
} catch (Exception $e) {
throw new Resque_RedisException('Error communicating with Redis: ' . $e->getMessage(), 0, $e);
}

56
lib/Resque/Worker.php
View File

@ -21,7 +21,7 @@ class Resque_Worker
/**
* @var array Array of all associated queues for this worker.
*/
private $queues = array();
private $queues = [];
/**
* @var string The hostname of this worker.
@ -69,7 +69,7 @@ class Resque_Worker
$this->logger = new Resque_Log();
if (!is_array($queues)) {
$queues = array($queues);
$queues = [$queues];
}
$this->queues = $queues;
@ -86,10 +86,10 @@ class Resque_Worker
{
$workers = Resque::redis()->smembers('workers');
if (!is_array($workers)) {
$workers = array();
$workers = [];
}
$instances = array();
$instances = [];
foreach ($workers as $workerId) {
$instances[] = self::find($workerId);
}
@ -158,7 +158,7 @@ class Resque_Worker
$job = false;
if (!$this->paused) {
if ($blocking === true) {
$this->logger->log(Psr\Log\LogLevel::INFO, 'Starting blocking with timeout of {interval}', array('interval' => $interval));
$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);
@ -175,7 +175,7 @@ class Resque_Worker
if ($blocking === false) {
// If no job was found, we sleep for $interval before continuing and checking again
$this->logger->log(Psr\Log\LogLevel::INFO, 'Sleeping for {interval}', array('interval' => $interval));
$this->logger->log(Psr\Log\LogLevel::INFO, 'Sleeping for {interval}', ['interval' => $interval]);
if ($this->paused) {
$this->updateProcLine('Paused');
} else {
@ -188,7 +188,7 @@ class Resque_Worker
continue;
}
$this->logger->log(Psr\Log\LogLevel::NOTICE, 'Starting work on {job}', array('job' => $job));
$this->logger->log(Psr\Log\LogLevel::NOTICE, 'Starting work on {job}', ['job' => $job]);
Resque_Event::trigger('beforeFork', $job);
$this->workingOn($job);
@ -239,13 +239,13 @@ class Resque_Worker
Resque_Event::trigger('afterFork', $job);
$job->perform();
} catch (Exception $e) {
$this->logger->log(Psr\Log\LogLevel::CRITICAL, '{job} has failed {stack}', array('job' => $job, 'stack' => $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', array('job' => $job));
$this->logger->log(Psr\Log\LogLevel::NOTICE, '{job} has finished', ['job' => $job]);
}
/**
@ -257,21 +257,21 @@ class Resque_Worker
{
$queues = $this->queues();
if (!is_array($queues)) {
return;
return false;
}
if ($blocking === true) {
$job = Resque_Job::reserveBlocking($queues, $timeout);
if ($job) {
$this->logger->log(Psr\Log\LogLevel::INFO, 'Found job on {queue}', array('queue' => $job->queue));
$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', array('queue' => $queue));
$this->logger->log(Psr\Log\LogLevel::INFO, 'Checking {queue} for jobs', ['queue' => $queue]);
$job = Resque_Job::reserve($queue);
if ($job) {
$this->logger->log(Psr\Log\LogLevel::INFO, 'Found job on {queue}', array('queue' => $job->queue));
$this->logger->log(Psr\Log\LogLevel::INFO, 'Found job on {queue}', ['queue' => $job->queue]);
return $job;
}
}
@ -344,12 +344,12 @@ class Resque_Worker
return;
}
pcntl_signal(SIGTERM, array($this, 'shutDownNow'));
pcntl_signal(SIGINT, array($this, 'shutDownNow'));
pcntl_signal(SIGQUIT, array($this, 'shutdown'));
pcntl_signal(SIGUSR1, array($this, 'killChild'));
pcntl_signal(SIGUSR2, array($this, 'pauseProcessing'));
pcntl_signal(SIGCONT, array($this, 'unPauseProcessing'));
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');
}
@ -403,13 +403,13 @@ class Resque_Worker
return;
}
$this->logger->log(Psr\Log\LogLevel::INFO, 'Killing child at {child}', array('child' => $this->child));
$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.', array('child' => $this->child));
$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.', array('child' => $this->child));
$this->logger->log(Psr\Log\LogLevel::INFO, 'Child {child} not found, restarting.', ['child' => $this->child]);
$this->shutdown();
}
}
@ -432,7 +432,7 @@ class Resque_Worker
if ($host != $this->hostname || in_array($pid, $workerPids) || $pid == getmypid()) {
continue;
}
$this->logger->log(Psr\Log\LogLevel::INFO, 'Pruning dead worker: {worker}', array('worker' => (string)$worker));
$this->logger->log(Psr\Log\LogLevel::INFO, 'Pruning dead worker: {worker}', ['worker' => (string)$worker]);
$worker->unregisterWorker();
}
}
@ -446,7 +446,7 @@ class Resque_Worker
*/
public function workerPids()
{
$pids = array();
$pids = [];
exec('ps -A -o pid,command | grep [r]esque', $cmdOutput);
foreach ($cmdOutput as $line) {
list($pids[],) = explode(' ', trim($line), 2);
@ -490,11 +490,11 @@ class Resque_Worker
$job->worker = $this;
$this->currentJob = $job;
$job->updateStatus(Resque_Job_Status::STATUS_RUNNING);
$data = json_encode(array(
$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);
}
@ -523,13 +523,13 @@ class Resque_Worker
/**
* Return an object describing the job this worker is currently working on.
*
* @return object Object with details of current job.
* @return array Array with details of current job.
*/
public function job()
{
$job = Resque::redis()->get('worker:' . $this);
if (!$job) {
return array();
return [];
} else {
return json_decode($job, true);
}