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

#218 Remove item from queue as well when args match

Browse Source
This commit is contained in:
Axel K. 2014-11-10 15:38:58 +01:00
parent 226ec33bb0
commit 6cda08de25
1 changed files with 32 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
lib/Resque.php
View File

@ -253,41 +253,42 @@ class Resque
*/
private static function removeItems($queue, $items = Array())
{
$counter = 0;
$originalQueue = 'queue:'. $queue;
$tempQueue = $originalQueue. ':temp:'. time();
$requeueQueue = $tempQueue. ':requeue';
$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)) {
$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;
}
/**