Add Reserver concept to allow different behaviours for reservation of jobs.

bin/resque

@@ -9,22 +9,25 @@ $files = array(
     __DIR__ . '/../vendor/autoload.php',
 );
 
-$found = false;
+$loader = null;
 foreach ($files as $file) {
     if (file_exists($file)) {
-        require_once $file;
+        $loader = require_once $file;
         break;
     }
 }
 
-if (!class_exists('Composer\Autoload\ClassLoader', false)) {
+if (!($loader instanceof Composer\Autoload\ClassLoader)) {
     die(
         'You need to set up the project dependencies using the following commands:' . PHP_EOL .
             'curl -s http://getcomposer.org/installer | php' . PHP_EOL .
             'php composer.phar install' . PHP_EOL
     );
 }
 
+use Resque\Reserver\ReserverFactory;
+use Resque\Reserver\UnknownReserverException;
+
 $QUEUE = getenv('QUEUE');
 if(empty($QUEUE)) {
     die("Set QUEUE env var containing the list of queues to work.\n");
@@ -41,10 +44,11 @@ $REDIS_BACKEND = getenv('REDIS_BACKEND');
 // A redis database number
 $REDIS_BACKEND_DB = getenv('REDIS_BACKEND_DB');
 if(!empty($REDIS_BACKEND)) {
-    if (empty($REDIS_BACKEND_DB))
+    if (empty($REDIS_BACKEND_DB)) {
         Resque::setBackend($REDIS_BACKEND);
-    else
+    } else {
         Resque::setBackend($REDIS_BACKEND, $REDIS_BACKEND_DB);
+    }
 }
 
 $logLevel = false;
@@ -67,13 +71,32 @@ if($APP_INCLUDE) {
     require_once $APP_INCLUDE;
 }
 
+// re-register the composer autoloader so that we use Resque here, in case APP_INCLUDE depends on a different version
+$loader->register(true);
+
 // See if the APP_INCLUDE containes a logger object,
 // If none exists, fallback to internal logger
 if (!isset($logger) || !is_object($logger)) {
     $logger = new Resque_Log($logLevel);
 }
 
-$BLOCKING = getenv('BLOCKING') !== FALSE;
+$reserverFactory = new ReserverFactory($logger);
+Resque_Worker::setReserverFactory($reserverFactory);
+
+$queues = explode(',', $QUEUE);
+if (!is_array($queues)) {
+    $queues = array($queues);
+}
+
+$reserver = null;
+try {
+    $reserver = $reserverFactory->createReserverFromEnvironment($queues);
+    $logger->notice('Using reserver {reserver}', array('reserver' => $reserver->getName()));
+} catch (UnknownReserverException $exception) {
+    $logger->emergency("Could not create reserver: {error}", ['error' => $exception->getMessage()]);
+    die;
+}
+
 
 $interval = 5;
 $INTERVAL = getenv('INTERVAL');
@@ -102,19 +125,17 @@ if($count > 1) {
         }
         // Child, start the worker
         else if(!$pid) {
-            $queues = explode(',', $QUEUE);
-            $worker = new Resque_Worker($queues);
+            $worker = new Resque_Worker($reserver, $queues);
             $worker->setLogger($logger);
             $logger->log(Psr\Log\LogLevel::NOTICE, 'Starting worker {worker}', array('worker' => $worker));
-            $worker->work($interval, $BLOCKING);
+            $worker->work($interval);
             break;
         }
     }
 }
 // Start a single worker
 else {
-    $queues = explode(',', $QUEUE);
-    $worker = new Resque_Worker($queues);
+    $worker = new Resque_Worker($reserver, $queues);
     $worker->setLogger($logger);
 
     $PIDFILE = getenv('PIDFILE');
@@ -124,6 +145,5 @@ else {
     }
 
     $logger->log(Psr\Log\LogLevel::NOTICE, 'Starting worker {worker}', array('worker' => $worker));
-    $worker->work($interval, $BLOCKING);
+    $worker->work($interval);
 }
-?>

composer.json

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

lib/Resque/Reserver/AbstractReserver.php

@@ -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, '\\');
+    }
+}