blocking pop from redis queues

laravel · Dec 2, 2017 · 63b6108 · 63b6108
1 parent 64ff115
commit 63b6108
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 2 deletions.
diff --git a/src/Illuminate/Queue/Connectors/RedisConnector.php b/src/Illuminate/Queue/Connectors/RedisConnector.php
@@ -45,7 +45,8 @@ public function connect(array $config)
         return new RedisQueue(
             $this->redis, $config['queue'],
             $config['connection'] ?? $this->connection,
-            $config['retry_after'] ?? 60
+            $config['retry_after'] ?? 60,
+            $config['block_for'] ?? 0
         );
     }
 }
diff --git a/src/Illuminate/Queue/RedisQueue.php b/src/Illuminate/Queue/RedisQueue.php
@@ -37,21 +37,30 @@ class RedisQueue extends Queue implements QueueContract
      */
     protected $retryAfter = 60;
 
+    /**
+     * The maximum number of seconds to block for a job.
+     *
+     * @var int
+     */
+    private $blockFor = 0;
+
     /**
      * Create a new Redis queue instance.
      *
      * @param  \Illuminate\Contracts\Redis\Factory  $redis
      * @param  string  $default
      * @param  string  $connection
      * @param  int  $retryAfter
+     * @param  int  $blockFor
      * @return void
      */
-    public function __construct(Redis $redis, $default = 'default', $connection = null, $retryAfter = 60)
+    public function __construct(Redis $redis, $default = 'default', $connection = null, $retryAfter = 60, $blockFor = 0)
     {
         $this->redis = $redis;
         $this->default = $default;
         $this->connection = $connection;
         $this->retryAfter = $retryAfter;
+        $this->blockFor = $blockFor;
     }
 
     /**
@@ -199,6 +208,45 @@ public function migrateExpiredJobs($from, $to)
      * @return array
      */
     protected function retrieveNextJob($queue)
+    {
+        if ($this->blockFor >= 1) {
+            return $this->blockingPop($queue);
+        }
+
+        return $this->nonBlockingPop($queue);
+    }
+
+    /**
+     * Retrieve the next job by blocking-pop.
+     *
+     * @param  string  $queue
+     * @return array
+     */
+    protected function blockingPop($queue)
+    {
+        $rawBody = $this->getConnection()->blpop($queue, $this->blockFor);
+
+        if (! is_null($rawBody)) {
+            $payload = json_decode($rawBody[1], true);
+            $payload['attempts']++;
+            $reserved = json_encode($payload);
+            $this->getConnection()->zadd($queue.':reserved', [
+                $reserved => $this->availableAt($this->retryAfter),
+            ]);
+
+            return [$rawBody[1], $reserved];
+        }
+
+        return [null, null];
+    }
+
+    /**
+     * Retrieve the next job by Lua script.
+     *
+     * @param  string  $queue
+     * @return mixed
+     */
+    protected function nonBlockingPop($queue)
     {
         return $this->getConnection()->eval(
             LuaScripts::pop(), 2, $queue, $queue.':reserved',