支持失败并重回队列操作

Author: Yurunsoft

README.md

@@ -147,6 +147,9 @@ if(null !== $message)
 
     // 将消息标记为失败
     $queue->fail($message);
+
+    // 将消息标记为失败，并重回队列
+    $queue->fail($message, true);
 }
 ```
 

src/Driver/IQueueDriver.php

@@ -61,9 +61,10 @@ public function success(IMessage $message);
      * 将消息标记为失败
      *
      * @param \Imi\Queue\Contract\IMessage $message
+     * @param bool $requeue
      * @return void
      */
-    public function fail(IMessage $message);
+    public function fail(IMessage $message, bool $requeue = false);
 
     /**
      * 获取队列状态

src/Driver/RedisQueueDriver.php

@@ -328,21 +328,35 @@ public function success(IMessage $message)
      * 将消息标记为失败
      *
      * @param \Imi\Queue\Contract\IMessage $message
+     * @param bool $requeue
      * @return void
      */
-    public function fail(IMessage $message)
+    public function fail(IMessage $message, bool $requeue = false)
     {
         $redis = RedisManager::getInstance($this->poolName);
+        if($requeue)
+        {
+            $operation = <<<LUA
+-- 加入队列
+redis.call('rpush', KEYS[2], ARGV[1]);
+LUA;
+        }
+        else
+        {
+            $operation = <<<LUA
+-- 加入失败队列
+redis.call('rpush', KEYS[2], ARGV[1])
+LUA;
+        }
         $result = $redis->evalEx(<<<LUA
 -- 从工作队列删除
 redis.call('zrem', KEYS[1], ARGV[1])
--- 加入失败队列
-redis.call('rpush', KEYS[2], ARGV[1])
+{$operation}
 return true
 LUA
         , [
             $this->getQueueKey(QueueType::WORKING),
-            $this->getQueueKey(QueueType::FAIL),
+            $requeue ? $this->getQueueKey(QueueType::READY) : $this->getQueueKey(QueueType::FAIL),
             $message->getMessageId(),
         ], 2);
 

tests/Queue/BaseQueueTest.php

@@ -81,9 +81,18 @@ public function testRestoreFailMessages()
         $message = $this->getDriver()->pop();
         $this->assertNotEmpty($message->getMessageId());
 
-        $driver->fail($message, 'gg');
+        $driver->fail($message);
 
         $this->assertEquals(1, $driver->restoreFailMessages());
+        
+        // requeue
+        $message = $this->getDriver()->pop();
+        $this->assertNotEmpty($message->getMessageId());
+
+        $driver->fail($message, true);
+
+        $this->assertEquals(1, $driver->status()->getReady());
+        $this->assertEquals(0, $driver->restoreFailMessages());
     }
 
     public function testRestoreTimeoutMessages()