Fixed bug that the older versions of parsers cannot parse new messages for async-queue.

Author: huangdijia

src/JobMessage.php

@@ -29,7 +29,12 @@ public function __serialize(): array
             $this->job = $this->job->compress();
         }
 
-        return [$this->job, $this->attempts];
+        return [
+            $this->job,  // Compatible with old version, will be removed at v3.2
+            $this->attempts,  // Compatible with old version, will be removed at v3.2
+            'job' => $this->job,
+            'attempts' => $this->attempts,
+        ];
     }
 
     public function __unserialize(array $data): void

tests/JobMessageTest.php

@@ -33,6 +33,9 @@ public function testJobMessageSerialize()
         );
 
         $serialized = $message->__serialize();
+
+        $this->assertEquals($serialized[0], $serialized['job']);
+        $this->assertEquals($serialized[1], $serialized['attempts']);
         $this->assertArrayHasKey('job', $serialized);
         $this->assertArrayHasKey('attempts', $serialized);
 
@@ -52,6 +55,17 @@ public function testJobMessageSerializeCompatible()
 
         $serialized = $message->__serialize();
 
+        $this->assertEquals($serialized[0], $serialized['job']);
+        $this->assertEquals($serialized[1], $serialized['attempts']);
+
+        $message = unserialize(serialize($message));
+        $this->assertInstanceOf(MessageInterface::class, $message);
+        $this->assertInstanceOf(JobInterface::class, $message->job());
+        $this->assertInstanceOf(JobInterface::class, $message->job());
+        $this->assertInstanceOf(DemoJob::class, $message->job());
+        $this->assertSame($id, $message->job()->id);
+        $this->assertSame(0, $message->getAttempts());
+
         $serialized = [
             'job' => $serialized['job'] ?? $serialized[0],
             'attempts' => 3,
@@ -75,4 +89,27 @@ public function testJobMessageSerializeCompatible()
         $this->assertSame($id, $message->job()->id);
         $this->assertSame(5, $message->getAttempts());
     }
+
+    public function testUnserializeAsOldJobMessage()
+    {
+        $id = rand(0, 9999);
+        $message = new JobMessage(
+            new DemoJob($id)
+        );
+
+        $serialized = serialize($message);
+        $serialized = str_replace(
+            sprintf('O:%d:"%s', strlen(\Hyperf\AsyncQueue\JobMessage::class), \Hyperf\AsyncQueue\JobMessage::class),
+            sprintf('O:%d:"%s', strlen(\HyperfTest\AsyncQueue\Stub\OldJobMessage::class), \HyperfTest\AsyncQueue\Stub\OldJobMessage::class),
+            $serialized
+        );
+        $message = unserialize($serialized);
+
+        $this->assertInstanceOf(MessageInterface::class, $message);
+        $this->assertInstanceOf(JobInterface::class, $message->job());
+        $this->assertInstanceOf(JobInterface::class, $message->job());
+        $this->assertInstanceOf(DemoJob::class, $message->job());
+        $this->assertSame($id, $message->job()->id);
+        $this->assertSame(0, $message->getAttempts());
+    }
 }

tests/RedisDriverTest.php

@@ -115,7 +115,7 @@ public function testAsyncQueueJobGenerate()
         $driver->push(new DemoJob($id, $model));
 
         $serialized = (string) Context::get('test.async-queue.lpush.value');
-        $this->assertSame(248, strlen($serialized));
+        $this->assertSame(264, strlen($serialized));
 
         /** @var JobMessage $class */
         $class = $packer->unpack($serialized);

tests/Stub/OldJobMessage.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace HyperfTest\AsyncQueue\Stub;
+
+use Hyperf\Contract\UnCompressInterface;
+
+class OldJobMessage extends \Hyperf\AsyncQueue\JobMessage
+{
+    public function __unserialize(array $data): void
+    {
+        [$job, $attempts] = $data;
+
+        if ($job instanceof UnCompressInterface) {
+            $job = $job->uncompress();
+        }
+
+        $this->job = $job;
+        $this->attempts = $attempts;
+    }
+}