Some refinements

This commit is contained in:
Holger Reinhardt 2016-07-29 09:17:13 +02:00
parent 7bb3c9ef85
commit b60fb0d712
4 changed files with 110 additions and 13 deletions

View File

@ -2,6 +2,7 @@
namespace Hlgrrnhrdt\Resque\Console; namespace Hlgrrnhrdt\Resque\Console;
use Illuminate\Console\Command; use Illuminate\Console\Command;
use Symfony\Component\Console\Input\InputOption;
/** /**
* WorkCommand * WorkCommand
@ -14,6 +15,22 @@ class WorkCommand extends Command
public function fire() public function fire()
{ {
// $queue = $this->option('queue');
$interval = $this->option('interval');
$count = $this->option('count');
} }
/**
* {@inheritdoc}
*/
protected function getOptions()
{
return [
['queue', null, InputOption::VALUE_IS_ARRAY & InputOption::VALUE_OPTIONAL, '', 'default'],
['interval', null, InputOption::VALUE_OPTIONAL, '', 5],
['count', null, InputOption::VALUE_OPTIONAL, '', 1],
];
}
} }

View File

@ -8,15 +8,20 @@ namespace Hlgrrnhrdt\Resque;
*/ */
abstract class Job abstract class Job
{ {
/**
* @var \Resque_Job
*/
public $job;
/** /**
* @var string * @var string
*/ */
protected $queue = 'default'; public $queue = 'default';
/** /**
* @var array * @var array
*/ */
protected $arguments = []; public $args = [];
/** /**
* @return mixed * @return mixed
@ -31,7 +36,7 @@ abstract class Job
*/ */
public function arguments() public function arguments()
{ {
return $this->arguments; return $this->$args;
} }
/** /**
@ -44,4 +49,6 @@ abstract class Job
$this->queue = $queue; $this->queue = $queue;
return $this; return $this;
} }
abstract public function perform();
} }

View File

@ -9,14 +9,32 @@ namespace Hlgrrnhrdt\Resque;
class Queue class Queue
{ {
protected $name; protected $name;
/**
* @var ResqueManager
*/
private $manager;
public function __construct($name) /**
* Queue constructor.
*
* @param ResqueManager $manager
*/
public function __construct(ResqueManager $manager)
{ {
$this->name = $name; $this->manager = $manager;
} }
/**
* @return \Resque_Job[]
*/
public function jobs() public function jobs()
{ {
\Resque::redis()->lrange('queue:' . $this->name, 0, -1); $result = $this->manager->redis()->lrange('queue:' . $this->name, 0, -1);
$jobs = [];
foreach ($result as $job) {
$jobs[] = new \Resque_Job($this->name, json_decode($job, true));
}
return $jobs;
} }
} }

View File

@ -1,7 +1,8 @@
<?php <?php
namespace Hlgrrnhrdt\Resque; namespace Hlgrrnhrdt\Resque;
use Resque as PhpResque; use Resque;
use RuntimeException;
/** /**
* Resque * Resque
@ -15,19 +16,73 @@ class ResqueManager
*/ */
private $resque; private $resque;
public function __construct(PhpResque $resque) public function __construct(Resque $resque)
{ {
$this->resque = $resque; $this->resque = $resque;
} }
public function enqueue(Job $job) /**
* @param Job $job
* @param bool $trackStatus
*
* @return null|\Resque_Job_Status
*/
public function enqueue(Job $job, $trackStatus = false)
{ {
$this->resque->enqueue($job->queue(), get_class($job), $job->arguments()); $id = $this->resque->enqueue($job->queue(), get_class($job), $job->arguments(), $trackStatus);
if (true === $trackStatus) {
return new \Resque_Job_Status($id);
} }
public function enqueueOnce(Job $job) return null;
}
/**
* @param Job $job
* @param bool $trackStatus
*
* @return null|\Resque_Job_Status
*/
public function enqueueOnce(Job $job, $trackStatus = false)
{ {
$queue = new Queue($job->queue()); $queue = new Queue($job->queue());
foreach ($queue->jobs() as $queuedJob) {
if ($job->payload['class'] === get_class($queuedJob) && count(array_intersect($queuedJob->getArguments(),
$job->arguments())) === $job->arguments()
) {
return ($trackStatus) ? new \Resque_Job_Status($job->payload['id']) : null;
}
}
return $this->enqueue($job, $trackStatus);
}
/**
* @return \Resque_Redis
*/
public function redis()
{
return $this->resque->redis();
}
/**
* @return int
*
* @throws RuntimeException
*/
public function fork()
{
if (false === function_exists('pcntl_fork')) {
return -1;
}
$pid = pcntl_fork();
if (-1 === $pid) {
throw new RuntimeException('Unable to fork child worker.');
}
return $pid;
} }
} }