Add QueueOrderReserver.

This reserver checks queues in the order they were specified.
As long as jobs exist on higher priority queues, they will be reserved
before moving to the next lowest priority queue.

This implements the current default job reservation behaviour.

Author: rayward

composer.json

@@ -37,5 +37,10 @@
 		"psr-0": {
 			"Resque": "lib"
 		}
+	},
+	"autoload-dev": {
+		"psr-0": {
+			"Resque": "test/"
+		}
 	}
 }

composer.lock

@@ -0,0 +1,59 @@
+<?php
+
+namespace Resque\Reserver;
+
+use Resque_Job;
+use Psr\Log\LoggerInterface;
+use Resque;
+
+abstract class AbstractReserver implements ReserverInterface
+{
+    /** @var array */
+    private $queues;
+
+    /** @var LoggerInterface */
+    protected $logger;
+
+    /**
+     * @param LoggerInterface $logger
+     * @param array $queues The queues to reserve from. If this contains '*', then the queues are retrieved dynamically
+     * from redis on each call to reserve().
+     */
+    public function __construct(LoggerInterface $logger, array $queues)
+    {
+        $this->logger = $logger;
+        $this->queues = $queues;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getQueues()
+    {
+        if (in_array('*', $this->queues)) {
+            $queues = Resque::queues();
+    		sort($queues);
+            return $queues;
+        }
+
+        return $this->queues;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function waitAfterReservationAttempt()
+    {
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getName()
+    {
+        $name = get_class($this);
+        $name = str_replace(__NAMESPACE__, '', $name);
+        return trim($name, '\\');
+    }
+}

lib/Resque/Reserver/AbstractReserver.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace Resque\Reserver;
+
+use Resque_Job;
+
+/**
+ * QueueOrderReserver reserves jobs in the order that the queues is given. As long as jobs exist in a higher priority
+ * queue, they will continue to be reserved before moving to the next lowest priority queue.
+ *
+ * For example: given queues A, B and C, all the jobs from queue A will be processed before moving onto queue B and
+ * then after that queue C.
+ *
+ * This is the default reserver.
+ */
+class QueueOrderReserver extends AbstractReserver implements ReserverInterface
+{
+    /**
+     * {@inheritDoc}
+     */
+    public function reserve()
+    {
+        foreach ($this->getQueues() as $queue) {
+            $this->logger->debug("[{reserver}] Checking queue '{queue}' for jobs", array(
+                'queue'    => $queue,
+                'reserver' => $this->getName(),
+            ));
+
+            $job = Resque_Job::reserve($queue);
+            if ($job) {
+                $this->logger->info("[{reserver}] Found job on queue '{queue}'", array(
+                    'queue'    => $queue,
+                    'reserver' => $this->getName(),
+                ));
+                return $job;
+            }
+        }
+
+        return null;
+    }
+}

lib/Resque/Reserver/QueueOrderReserver.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace Resque\Tests\Reserver;
+
+use Resque\Reserver\ReserverInterface;
+use Resque;
+
+abstract class AbstractReserverTest extends \Resque_Tests_TestCase
+{
+    /** @var string */
+    protected $reserverName;
+
+    /**
+     * Gets a reserver instance configured with the given queues.
+     *
+     * @param array $queues
+     * @return ReserverInstance
+     */
+    abstract protected function getReserver(array $queues = array());
+
+    public function testGetName()
+    {
+        $this->assertEquals($this->reserverName, $this->getReserver()->getName());
+    }
+
+    public function testGetQueuesReturnsConfiguredQueues()
+    {
+        $queues = array(
+            'queue_' . rand(1, 100),
+            'queue_' . rand(101, 200),
+            'queue_' . rand(201, 300),
+        );
+        $this->assertEquals($queues, $this->getReserver($queues)->getQueues());
+    }
+
+    public function testGetQueuesWithAsterixQueueReturnsAllQueuesFromRedisInSortedOrder()
+    {
+        $queues = array(
+            'queue_b',
+            'queue_c',
+            'queue_d',
+            'queue_a',
+        );
+
+        // register queues in redis
+        foreach ($queues as $queue) {
+            Resque::redis()->sadd('queues', $queue);
+        }
+
+        $expected = array(
+            'queue_a',
+            'queue_b',
+            'queue_c',
+            'queue_d',
+        );
+
+        $this->assertEquals($expected, $this->getReserver(array('*'))->getQueues());
+    }
+}