Fixed bug that async-queue broken caused by uncompress model failed.

Author: limingxinleo

composer.json

@@ -18,7 +18,7 @@
         "php": ">=7.2",
         "psr/container": "^1.0|^2.0",
         "psr/event-dispatcher": "^1.0",
-        "hyperf/contract": "~2.2.0",
+        "hyperf/contract": "~2.2.8",
         "hyperf/command": "~2.2.0",
         "hyperf/utils": "~2.2.0"
     },

src/Driver/Driver.php

@@ -19,6 +19,7 @@
 use Hyperf\AsyncQueue\Exception\InvalidPackerException;
 use Hyperf\AsyncQueue\MessageInterface;
 use Hyperf\Contract\PackerInterface;
+use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Process\ProcessManager;
 use Hyperf\Utils\Arr;
 use Hyperf\Utils\Coroutine\Concurrent;
@@ -81,29 +82,34 @@ public function consume(): void
         $maxMessages = Arr::get($this->config, 'max_messages', 0);
 
         while (ProcessManager::isRunning()) {
-            [$data, $message] = $this->pop();
+            try {
+                [$data, $message] = $this->pop();
 
-            if ($data === false) {
-                continue;
-            }
+                if ($data === false) {
+                    continue;
+                }
 
-            $callback = $this->getCallback($data, $message);
+                $callback = $this->getCallback($data, $message);
 
-            if ($this->concurrent instanceof Concurrent) {
-                $this->concurrent->create($callback);
-            } else {
-                parallel([$callback]);
-            }
+                if ($this->concurrent instanceof Concurrent) {
+                    $this->concurrent->create($callback);
+                } else {
+                    parallel([$callback]);
+                }
 
-            if ($messageCount % $this->lengthCheckCount === 0) {
-                $this->checkQueueLength();
-            }
+                if ($messageCount % $this->lengthCheckCount === 0) {
+                    $this->checkQueueLength();
+                }
 
-            if ($maxMessages > 0 && $messageCount >= $maxMessages) {
-                break;
+                if ($maxMessages > 0 && $messageCount >= $maxMessages) {
+                    break;
+                }
+            } catch (\Throwable $exception) {
+                $logger = $this->container->get(StdoutLoggerInterface::class);
+                $logger->error((string) $exception);
+            } finally {
+                ++$messageCount;
             }
-
-            ++$messageCount;
         }
     }
 

tests/RedisDriverTest.php

@@ -83,6 +83,22 @@ public function testDemoModelGenerate()
         $this->assertEquals($model, $model2);
     }
 
+    public function testDemoModelUnCompressToNull()
+    {
+        $content = Str::random(1000);
+
+        $model = new DemoModel(9999, 'Hyperf', 1, $content);
+        $s1 = serialize($model);
+        $this->assertSame(1131, strlen($s1));
+
+        $meta = $model->compress();
+        $s2 = serialize($meta);
+        $this->assertSame(68, strlen($s2));
+        $this->assertInstanceOf(DemoModelMeta::class, $meta);
+
+        $this->assertNull($meta->uncompress());
+    }
+
     public function testAsyncQueueJobGenerate()
     {
         $container = $this->getContainer();

tests/Stub/DemoModelMeta.php

@@ -11,7 +11,6 @@
  */
 namespace HyperfTest\AsyncQueue\Stub;
 
-use Hyperf\Contract\CompressInterface;
 use Hyperf\Contract\UnCompressInterface;
 use Hyperf\Utils\Context;
 
@@ -24,10 +23,14 @@ public function __construct($id)
         $this->id = $id;
     }
 
-    public function uncompress(): CompressInterface
+    public function uncompress()
     {
         $data = Context::get('test.async-queue.demo.model.' . $this->id);
 
+        if ($this->id === 9999) {
+            return null;
+        }
+
         return new DemoModel($this->id, ...$data);
     }
 }