Basic support for blocking list pop

Author: ruudk

lib/Resque.php

@@ -14,6 +14,8 @@ class Resque
 {
 	const VERSION = '1.0';
 
+    const DEFAULT_INTERVAL = 5;
+
 	/**
 	 * @var Resque_Redis Instance of Resque_Redis that talks to redis.
 	 */
@@ -115,14 +117,19 @@ public static function push($queue, $item)
 	 * @param string $queue The name of the queue to fetch an item from.
 	 * @return array Decoded item from the queue.
 	 */
-	public static function pop($queue)
+	public static function pop($queue, $interval = null)
 	{
-		$item = self::redis()->lpop('queue:' . $queue);
+        if($interval == null) {
+            $item = self::redis()->lpop('queue:' . $queue);
+        } else {
+            $item = self::redis()->blpop('queue:' . $queue, $interval ?: Resque::DEFAULT_INTERVAL);
+        }
+
 		if(!$item) {
 			return;
 		}
 
-		return json_decode($item, true);
+		return json_decode($interval == 0 ? $item : $item[1], true);
 	}
 
 	/**
@@ -168,10 +175,10 @@ public static function enqueue($queue, $class, $args = null, $trackStatus = fals
 	 * @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)
+	public static function reserve($queue, $interval = null)
 	{
 		require_once dirname(__FILE__) . '/Resque/Job.php';
-		return Resque_Job::reserve($queue);
+		return Resque_Job::reserve($queue, $interval);
 	}
 
 	/**

lib/Resque/Job.php

@@ -83,9 +83,9 @@ public static function create($queue, $class, $args = null, $monitor = false)
 	 * @param string $queue The name of the queue to check for a job in.
 	 * @return null|object Null when there aren't any waiting jobs, instance of Resque_Job when a job was found.
 	 */
-	public static function reserve($queue)
+	public static function reserve($queue, $interval = null)
 	{
-		$payload = Resque::pop($queue);
+		$payload = Resque::pop($queue, $interval);
 		if(!is_array($payload)) {
 			return false;
 		}
@@ -156,7 +156,7 @@ public function getInstance()
 			);
 		}
 
-		$this->instance = new $this->payload['class']();
+		$this->instance = new $this->payload['class'];
 		$this->instance->job = $this;
 		$this->instance->args = $this->getArguments();
 		$this->instance->queue = $this->queue;

lib/Resque/Redis.php

@@ -50,6 +50,7 @@ class Resque_Redis extends Redisent
 		'lset',
 		'lrem',
 		'lpop',
+		'blpop',
 		'rpop',
 		'sadd',
 		'srem',

lib/Resque/RedisCluster.php

@@ -50,6 +50,7 @@ class Resque_RedisCluster extends RedisentCluster
 		'lset',
 		'lrem',
 		'lpop',
+		'blpop',
 		'rpop',
 		'sadd',
 		'srem',

lib/Resque/Worker.php

@@ -153,7 +153,7 @@ public function __construct($queues)
 	 *
 	 * @param int $interval How often to check for new jobs across the queues.
 	 */
-	public function work($interval = 5)
+	public function work($interval = Resque::DEFAULT_INTERVAL)
 	{
 		$this->updateProcLine('Starting');
 		$this->startup();
@@ -166,25 +166,25 @@ public function work($interval = 5)
 			// Attempt to find and reserve a job
 			$job = false;
 			if(!$this->paused) {
-				$job = $this->reserve();
+				$job = $this->reserve($interval);
 			}
 
-			if(!$job) {
-				// For an interval of 0, break now - helps with unit testing etc
-				if($interval == 0) {
-					break;
-				}
-				// If no job was found, we sleep for $interval before continuing and checking again
-				$this->log('Sleeping for ' . $interval, true);
-				if($this->paused) {
-					$this->updateProcLine('Paused');
-				}
-				else {
-					$this->updateProcLine('Waiting for ' . implode(',', $this->queues));
-				}
-				usleep($interval * 1000000);
-				continue;
-			}
+            $this->updateProcLine('Waiting for ' . implode(',', $this->queues) . ' with interval ' . $interval);
+
+            if(!$job) {
+                // For an interval of 0, break now - helps with unit testing etc
+                if($interval == 0) {
+                    break;
+                }
+
+                // If no job was found, we sleep for $interval before continuing and checking again
+                if($this->paused) {
+                    $this->updateProcLine('Paused');
+                    usleep($interval * 1000000); //it's paused, so don't hog redis with requests.
+                }
+
+                continue;
+            }
 
 			$this->log('got ' . $job);
 			Resque_Event::trigger('beforeFork', $job);
@@ -252,15 +252,15 @@ public function perform(Resque_Job $job)
 	 *
 	 * @return object|boolean Instance of Resque_Job if a job is found, false if not.
 	 */
-	public function reserve()
+	public function reserve($interval = 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);
+			$this->log('Checking ' . $queue . ' with interval ' . $interval, self::LOG_VERBOSE);
+			$job = Resque_Job::reserve($queue, $interval);
 			if($job) {
 				$this->log('Found job on ' . $queue, self::LOG_VERBOSE);
 				return $job;