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

Working blocking list pop :)

Browse source
This commit is contained in:
Ruud Kamphuis 2013-03-13 12:41:32 +01:00
parent 132531e1a2
commit c2c4d06f7b
14 changed files with 606 additions and 177 deletions
Download patch file Download diff file Expand all files Collapse all files

17
lib/Resque.php
View file

@ -122,20 +122,21 @@ class Resque
}
/**
* Pop an item off the end of the specified queue, decode it and
* return it.
* Pop an item off the end of the specified queues, using blocking list pop,
* 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.
* @param array $queues
* @param int $timeout
* @return null|array Decoded item from the queue.
*/
public static function blpop($queues, $interval = null)
public static function blpop(array $queues, $timeout)
{
$list = array();
foreach($queues AS $queue) {
$list[] = 'queue:' . $queue;
}
$item = self::redis()->blpop($list, $interval ? (int)$interval : Resque::DEFAULT_INTERVAL);
$item = self::redis()->blpop($list, (int)$timeout);
if(!$item) {
return;
@ -186,9 +187,9 @@ class Resque
* @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, $interval = null)
public static function reserve($queue)
{
return Resque_Job::reserve($queue, $interval);
return Resque_Job::reserve($queue);
}
/**

13
lib/Resque/Job.php
View file

@ -89,21 +89,20 @@ class Resque_Job
}
/**
* Find the next available job from the specified queue and return an
* instance of Resque_Job for it.
* Find the next available job from the specified queues using blocking list pop
* and return an instance of Resque_Job for it.
*
* @param string $queue The name of the queue to check for a job in.
* @param array $queues
* @param int $timeout
* @return null|object Null when there aren't any waiting jobs, instance of Resque_Job when a job was found.
*/
public static function reserveBlocking($queues, $interval = null)
public static function reserveBlocking(array $queues, $timeout = null)
{
$payload = Resque::blpop($queues, $interval);
$payload = Resque::blpop($queues, $timeout);
if(!is_array($payload)) {
return false;
}
var_dump($payload);
return new Resque_Job($payload->queue, $payload);
}

99
lib/Resque/Worker.php
View file

@ -147,13 +147,52 @@ class Resque_Worker
*
* @param int $interval How often to check for new jobs across the queues.
*/
public function work($interval = Resque::DEFAULT_INTERVAL)
public function work($interval = Resque::DEFAULT_INTERVAL, $blocking = false)
{
$this->updateProcLine('Starting');
$this->startup();
while($job = $this->reserveBlocking($interval)) {
$this->updateProcLine('Waiting for ' . implode(',', $this->queues) . ' with interval ' . $interval);
while(true) {
if($this->shutdown) {
break;
}
// Attempt to find and reserve a job
$job = false;
if(!$this->paused) {
if($blocking === true) {
$this->log('Starting blocking with timeout of ' . $interval, self::LOG_VERBOSE);
$this->updateProcLine('Waiting for ' . implode(',', $this->queues) . ' with blocking timeout ' . $interval);
} else {
$this->log('Starting with interval of ' . $interval, self::LOG_VERBOSE);
$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->log('Sleeping for ' . $interval, self::LOG_VERBOSE);
if($this->paused) {
$this->updateProcLine('Paused');
}
else {
$this->updateProcLine('Waiting for ' . implode(',', $this->queues));
}
usleep($interval * 1000000);
}
continue;
}
$this->log('got ' . $job);
Resque_Event::trigger('beforeFork', $job);
@ -190,14 +229,6 @@ class Resque_Worker
$this->child = null;
$this->doneWorking();
if($this->shutdown) {
break;
}
if($this->paused) {
break;
}
}
$this->unregisterWorker();
@ -225,44 +256,32 @@ class Resque_Worker
}
/**
* Attempt to find a job from the top of one of the queues for this worker.
*
* @return object|boolean Instance of Resque_Job if a job is found, false if not.
* @param bool $blocking
* @param int $timeout
* @return object|boolean Instance of Resque_Job if a job is found, false if not.
*/
public function reserve()
public function reserve($blocking = false, $timeout = null)
{
$queues = $this->queues();
if(!is_array($queues)) {
return;
}
foreach($queues as $queue) {
$this->log('Checking ' . $queue, self::LOG_VERBOSE);
$job = Resque_Job::reserve($queue);
if($blocking === true) {
$job = Resque_Job::reserveBlocking($queues, $timeout);
if($job) {
$this->log('Found job on ' . $queue, self::LOG_VERBOSE);
$this->log('Found job on ' . $job->queue, self::LOG_VERBOSE);
return $job;
}
}
return false;
}
/**
* Attempt to find a job from the top of one of the queues for this worker.
*
* @return object|boolean Instance of Resque_Job if a job is found, false if not.
*/
public function reserveBlocking($interval = null)
{
$queues = $this->queues();
if(!is_array($queues)) {
return;
}
$job = Resque_Job::reserveBlocking($queues, $interval);
if($job) {
$this->log('Found job on ' . $job->queue, self::LOG_VERBOSE);
return $job;
} else {
foreach($queues as $queue) {
$this->log('Checking ' . $queue, self::LOG_VERBOSE);
$job = Resque_Job::reserve($queue);
if($job) {
$this->log('Found job on ' . $queue, self::LOG_VERBOSE);
return $job;
}
}
}
return false;