idanoo/php-resque
1
0
Fork 0
mirror of https://github.com/idanoo/php-resque.git synced 2024-11-22 08:15:14 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #223 from theaxel/master

Browse Source 
Fix dequeueing with args
This commit is contained in:
Chris Boulton 2015-02-02 12:34:14 -08:00
commit 1d24d009f2
2 changed files with 55 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()) private static function removeItems($queue, $items = Array())
{ {
$counter = 0; $counter = 0;
$originalQueue = 'queue:'. $queue; $originalQueue = 'queue:'. $queue;
$tempQueue = $originalQueue. ':temp:'. time(); $tempQueue = $originalQueue. ':temp:'. time();
$requeueQueue = $tempQueue. ':requeue'; $requeueQueue = $tempQueue. ':requeue';
// move each item from original queue to temp queue and process it // move each item from original queue to temp queue and process it
$finished = false; $finished = false;
while(!$finished) { while (!$finished) {
$string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue); $string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue);
if(!empty($string)) { if (!empty($string)) {
if(self::matchItem($string, $items)) { if(self::matchItem($string, $items)) {
$counter++; self::redis()->rpop($tempQueue);
} else { $counter++;
self::redis()->rpoplpush($tempQueue, self::redis()->getPrefix() . $requeueQueue); } else {
} self::redis()->rpoplpush($tempQueue, self::redis()->getPrefix() . $requeueQueue);
} else { }
$finished = true; } else {
$finished = true;
}
} }
}
// move back from temp queue to original queue // move back from temp queue to original queue
$finished = false; $finished = false;
while(!$finished) { while (!$finished) {
$string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() .$originalQueue); $string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() .$originalQueue);
if (empty($string)) { if (empty($string)) {
$finished = true; $finished = true;
}
} }
}
// remove temp queue and requeue queue // remove temp queue and requeue queue
self::redis()->del($requeueQueue); self::redis()->del($requeueQueue);
self::redis()->del($tempQueue); self::redis()->del($tempQueue);
return $counter; return $counter;
} }
/** /**

23
test/Resque/Tests/JobTest.php
View File

@ -299,6 +299,29 @@ class Resque_Tests_JobTest extends Resque_Tests_TestCase
#$this->assertEquals(Resque::size($queue), 1); #$this->assertEquals(Resque::size($queue), 1);
} }
public function testDequeueSeveralItemsWithArgs()
{
// GIVEN
$queue = 'jobs';
$args = array('foo' => 1, 'bar' => 10);
$removeArgs = array('foo' => 1, 'bar' => 2);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $args);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $removeArgs);
Resque::enqueue($queue, 'Test_Job_Dequeue9', $removeArgs);
$this->assertEquals(Resque::size($queue), 3);
// WHEN
$test = array('Test_Job_Dequeue9' => $removeArgs);
$removedItems = Resque::dequeue($queue, $test);
// THEN
$this->assertEquals($removedItems, 2);
$this->assertEquals(Resque::size($queue), 1);
$item = Resque::pop($queue);
$this->assertInternalType('array', $item['args']);
$this->assertEquals(10, $item['args'][0]['bar'], 'Wrong items were dequeued from queue!');
}
public function testDequeueItemWithUnorderedArg() public function testDequeueItemWithUnorderedArg()
{ {
$queue = 'jobs'; $queue = 'jobs';