From a4d25775a23d0ac3fce95fe120aa5b78ba64f722 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Thu, 9 Jul 2026 10:44:00 +0300 Subject: [PATCH 1/4] Update for queue message API changes --- src/Adapter.php | 4 +- src/Message/Message.php | 42 ++++++++++++------- src/QueueProvider.php | 10 ++++- src/QueueProviderInterface.php | 5 ++- tests/Integration/QueueProviderTest.php | 2 +- tests/Integration/QueueTest.php | 11 ++--- .../Support/ExtendedSimpleMessageHandler.php | 2 +- tests/Support/IntegrationTestCase.php | 7 ++-- tests/Unit/Message/MessageTest.php | 23 +++++----- tests/Unit/QueueTest.php | 7 ++-- 10 files changed, 69 insertions(+), 44 deletions(-) diff --git a/src/Adapter.php b/src/Adapter.php index 0ce544c..707b5a0 100644 --- a/src/Adapter.php +++ b/src/Adapter.php @@ -9,7 +9,7 @@ use Yiisoft\Queue\Cli\LoopInterface; use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\Message\MessageInterface; -use Yiisoft\Queue\Message\MessageSerializerInterface; +use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface; use Yiisoft\Queue\MessageStatus; final class Adapter implements AdapterInterface @@ -63,7 +63,7 @@ public function status(int|string $id): MessageStatus public function push(MessageInterface $message): MessageInterface { $payload = $this->serializer->serialize($message); - $id = $this->provider->pushMessage($payload, $message->getMetadata()); + $id = $this->provider->pushMessage($payload, $message->getMeta()); return new IdEnvelope($message, $id); } diff --git a/src/Message/Message.php b/src/Message/Message.php index 93c9491..c7bbd03 100644 --- a/src/Message/Message.php +++ b/src/Message/Message.php @@ -4,28 +4,34 @@ namespace Yiisoft\Queue\Redis\Message; +use Yiisoft\Queue\Message\DelayEnvelope; use Yiisoft\Queue\Message\MessageInterface; +/** + * @psalm-import-type MessagePayload from MessageInterface + * @psalm-import-type MessageMeta from MessageInterface + */ final class Message implements MessageInterface { /** - * @param array $metadata + * @psalm-param MessagePayload $payload + * @psalm-param MessageMeta $meta */ public function __construct( private string $handlerName, - private mixed $data, - private array $metadata, + private bool|int|float|string|array|null $payload, + private array $meta, private int $delay = 0 //delay in seconds ) { if ($this->delay > 0) { - $this->metadata['delay'] = $delay; + $this->meta[DelayEnvelope::META_DELAY_SECONDS] = $delay; } } public function withDelay(int $delay): self { $message = clone $this; - $message->metadata['delay'] = $delay; + $message->meta[DelayEnvelope::META_DELAY_SECONDS] = $delay; return $message; } @@ -39,31 +45,37 @@ public function getType(): string return $this->handlerName; } - public function getData(): mixed + /** + * @psalm-return MessagePayload + */ + public function getPayload(): bool|int|float|string|array|null { - return $this->data; + return $this->payload; } /** - * @return array + * @psalm-return MessageMeta */ - public function getMetadata(): array + public function getMeta(): array { - return $this->metadata; + return $this->meta; } /** - * @param array $metadata + * @psalm-param MessageMeta $meta */ - public function withMetadata(array $metadata): static + public function withMeta(array $meta): static { $message = clone $this; - $message->metadata = $metadata; + $message->meta = $meta; return $message; } - public static function fromData(string $type, mixed $data): self + /** + * @psalm-param MessagePayload $payload + */ + public static function fromPayload(string $type, bool|int|float|string|array|null $payload): static { - return new self($type, $data, []); + return new self($type, $payload, []); } } diff --git a/src/QueueProvider.php b/src/QueueProvider.php index da397a8..1978cde 100644 --- a/src/QueueProvider.php +++ b/src/QueueProvider.php @@ -5,6 +5,7 @@ namespace Yiisoft\Queue\Redis; use RedisException; +use Yiisoft\Queue\Message\DelayEnvelope; use Yiisoft\Queue\Redis\Exception\NotConnectedRedisException; class QueueProvider implements QueueProviderInterface @@ -28,13 +29,18 @@ public function __construct( /** * @throws RedisException */ - public function pushMessage(string $message, array $metadata = []): int + /** + * @param array $meta + */ + public function pushMessage(string $message, array $meta = []): int { $this->checkConnection(); $id = $this->getId(); $this->redis->hset("$this->channelName.messages", (string) $id, $message); - $delay = isset($metadata['delay']) && is_int($metadata['delay']) ? $metadata['delay'] : 0; + /** @var bool|int|float|string|array|null $delay */ + $delay = $meta[DelayEnvelope::META_DELAY_SECONDS] ?? 0; + $delay = is_int($delay) || is_float($delay) ? $delay : 0; if ($delay > 0) { $this->redis->zadd("$this->channelName.delayed", time() + $delay, $id); } else { diff --git a/src/QueueProviderInterface.php b/src/QueueProviderInterface.php index 4efefd9..c911820 100644 --- a/src/QueueProviderInterface.php +++ b/src/QueueProviderInterface.php @@ -6,7 +6,10 @@ interface QueueProviderInterface { - public function pushMessage(string $message, array $metadata = []): int; + /** + * @param array $meta + */ + public function pushMessage(string $message, array $meta = []): int; /** * @return Reserve|null Payload and id, or null if queue is empty. diff --git a/tests/Integration/QueueProviderTest.php b/tests/Integration/QueueProviderTest.php index 09d2281..1b529b5 100644 --- a/tests/Integration/QueueProviderTest.php +++ b/tests/Integration/QueueProviderTest.php @@ -36,7 +36,7 @@ public function test__construct(): QueueProvider public function testDelay(QueueProvider $provider): void { $message = new Message('test', ['key' => 'value'], [], 2); - $id = $provider->pushMessage(json_encode($message->getData(), JSON_THROW_ON_ERROR), $message->getMetadata()); + $id = $provider->pushMessage(json_encode($message->getPayload(), JSON_THROW_ON_ERROR), $message->getMeta()); $this->assertGreaterThan(0, $id); $reserv = $provider->reserve($id); $this->assertNull($reserv); diff --git a/tests/Integration/QueueTest.php b/tests/Integration/QueueTest.php index 5683720..7a9b44b 100644 --- a/tests/Integration/QueueTest.php +++ b/tests/Integration/QueueTest.php @@ -6,10 +6,11 @@ use Yiisoft\Queue\Adapter\AdapterInterface; use Yiisoft\Queue\Cli\LoopInterface; -use Yiisoft\Queue\Message\JsonMessageSerializer; +use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder; +use Yiisoft\Queue\Message\Serializer\MessageSerializer; use Yiisoft\Queue\Message\GenericMessage as Message; use Yiisoft\Queue\Message\MessageInterface; -use Yiisoft\Queue\Message\MessageSerializerInterface; +use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface; use Yiisoft\Queue\MessageStatus; use Yiisoft\Queue\Queue; use Yiisoft\Queue\Redis\Adapter; @@ -65,7 +66,7 @@ public function testStatus(): void $mockReserved = $this->createMock(QueueProviderInterface::class); $mockReserved->method('existInReserved')->willReturn(true); - $adapter = new Adapter($mockReserved, new JsonMessageSerializer(), $this->getLoop()); + $adapter = new Adapter($mockReserved, new MessageSerializer(new JsonMessageEncoder()), $this->getLoop()); $status = $adapter->status('1'); $this->assertEquals(MessageStatus::RESERVED, $status); @@ -83,7 +84,7 @@ public function testListen(): void $adapter = new Adapter( $queueProvider ->withChannelName('yii-queue'), - new JsonMessageSerializer(), + new MessageSerializer(new JsonMessageEncoder()), $mockLoop, ); $queue = $this->getDefaultQueue($adapter); @@ -131,7 +132,7 @@ public function testAdapterNullMessage() $adapter = new Adapter( $provider, - new JsonMessageSerializer(), + new MessageSerializer(new JsonMessageEncoder()), $mockLoop, ); $notUseHandler = true; diff --git a/tests/Support/ExtendedSimpleMessageHandler.php b/tests/Support/ExtendedSimpleMessageHandler.php index ef74fc9..1cc87ab 100644 --- a/tests/Support/ExtendedSimpleMessageHandler.php +++ b/tests/Support/ExtendedSimpleMessageHandler.php @@ -17,7 +17,7 @@ public function __construct(private FileHelper $fileHelper) public function handle(MessageInterface $message): void { - $data = $message->getData(); + $data = $message->getPayload(); if (null !== $data) { $this->fileHelper->put($data['file_name'], $data['payload']['time']); } diff --git a/tests/Support/IntegrationTestCase.php b/tests/Support/IntegrationTestCase.php index 7eb8b4f..026b87c 100644 --- a/tests/Support/IntegrationTestCase.php +++ b/tests/Support/IntegrationTestCase.php @@ -11,7 +11,8 @@ use Yiisoft\Queue\Adapter\AdapterInterface; use Yiisoft\Queue\Cli\LoopInterface; use Yiisoft\Queue\Cli\SignalLoop; -use Yiisoft\Queue\Message\JsonMessageSerializer; +use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder; +use Yiisoft\Queue\Message\Serializer\MessageSerializer; use Yiisoft\Queue\Message\MessageInterface; use Yiisoft\Queue\Middleware\CallableFactory; use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher; @@ -91,7 +92,7 @@ protected function getMessageHandlers(): array return [ 'ext-simple' => [new ExtendedSimpleMessageHandler(new FileHelper()), 'handle'], 'exception-listen' => static function (MessageInterface $message) { - $data = $message->getData(); + $data = $message->getPayload(); if (null !== $data) { throw new \RuntimeException((string) $data['payload']['time']); } @@ -134,7 +135,7 @@ protected function getAdapter(): AdapterInterface { return $this->adapter ??= new Adapter( $this->getQueueProvider(), - new JsonMessageSerializer(), + new MessageSerializer(new JsonMessageEncoder()), $this->getLoop(), ); } diff --git a/tests/Unit/Message/MessageTest.php b/tests/Unit/Message/MessageTest.php index c1c3681..451799f 100644 --- a/tests/Unit/Message/MessageTest.php +++ b/tests/Unit/Message/MessageTest.php @@ -5,6 +5,7 @@ namespace Unit\Message; use PHPUnit\Framework\TestCase; +use Yiisoft\Queue\Message\DelayEnvelope; use Yiisoft\Queue\Redis\Message\Message; class MessageTest extends TestCase @@ -19,28 +20,28 @@ public function testGetHandlerName(): void public function testGetData(): void { $message = new Message('handler', 'data', []); - $this->assertEquals('data', $message->getData()); + $this->assertEquals('data', $message->getPayload()); } public function testGetMetadata(): void { $metadata = ['key' => 'value']; $message = new Message('handler', 'data', $metadata); - $this->assertEquals($metadata, $message->getMetadata()); + $this->assertEquals($metadata, $message->getMeta()); $message = new Message('handler', 'data', $metadata, 2); - $metadata['delay'] = 2; - $this->assertEquals($metadata, $message->getMetadata()); + $metadata[DelayEnvelope::META_DELAY_SECONDS] = 2; + $this->assertEquals($metadata, $message->getMeta()); } public function testWithMetadata(): void { $message = new Message('handler', 'data', []); - $messageWithMetadata = $message->withMetadata(['key' => 'value']); + $messageWithMetadata = $message->withMeta(['key' => 'value']); $this->assertNotSame($message, $messageWithMetadata); - $this->assertSame([], $message->getMetadata()); - $this->assertSame(['key' => 'value'], $messageWithMetadata->getMetadata()); + $this->assertSame([], $message->getMeta()); + $this->assertSame(['key' => 'value'], $messageWithMetadata->getMeta()); } public function testWithDelay(): void @@ -49,7 +50,7 @@ public function testWithDelay(): void $delayedMessage = $message->withDelay(5); $this->assertNotSame($message, $delayedMessage); - $this->assertEquals(5, $delayedMessage->getMetadata()['delay']); + $this->assertEquals(5, $delayedMessage->getMeta()[DelayEnvelope::META_DELAY_SECONDS]); } public function testFromData(): void @@ -57,11 +58,11 @@ public function testFromData(): void $handlerName = 'test-handler'; $data = ['key' => 'value']; - $message = Message::fromData($handlerName, $data); + $message = Message::fromPayload($handlerName, $data); $this->assertInstanceOf(Message::class, $message); $this->assertEquals($handlerName, $message->getHandlerName()); - $this->assertEquals($data, $message->getData()); - $this->assertEquals([], $message->getMetadata()); + $this->assertEquals($data, $message->getPayload()); + $this->assertEquals([], $message->getMeta()); } } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index abbee71..cfa4712 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -6,10 +6,11 @@ use PHPUnit\Framework\TestCase; use Yiisoft\Queue\Cli\LoopInterface; -use Yiisoft\Queue\Message\JsonMessageSerializer; +use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder; +use Yiisoft\Queue\Message\Serializer\MessageSerializer; use Yiisoft\Queue\Message\Message; use Yiisoft\Queue\Message\MessageInterface; -use Yiisoft\Queue\Message\MessageSerializerInterface; +use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface; use Yiisoft\Queue\Redis\Adapter; use Yiisoft\Queue\Redis\QueueProviderInterface; @@ -37,7 +38,7 @@ public function testAdapterNullMessage() $adapter = new Adapter( $provider, - new JsonMessageSerializer(), + new MessageSerializer(new JsonMessageEncoder()), $mockLoop, ); $notUseHandler = true; From 6ec0afe7a364ead88e192d4365fffe65bfe97a8e Mon Sep 17 00:00:00 2001 From: StyleCI Bot Date: Thu, 9 Jul 2026 08:08:39 +0000 Subject: [PATCH 2/4] Apply fixes from StyleCI --- src/QueueProvider.php | 4 ++-- src/QueueProviderInterface.php | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/QueueProvider.php b/src/QueueProvider.php index 1978cde..38e37fb 100644 --- a/src/QueueProvider.php +++ b/src/QueueProvider.php @@ -30,7 +30,7 @@ public function __construct( * @throws RedisException */ /** - * @param array $meta + * @param array $meta */ public function pushMessage(string $message, array $meta = []): int { @@ -38,7 +38,7 @@ public function pushMessage(string $message, array $meta = []): int $id = $this->getId(); $this->redis->hset("$this->channelName.messages", (string) $id, $message); - /** @var bool|int|float|string|array|null $delay */ + /** @var array|bool|float|int|string|null $delay */ $delay = $meta[DelayEnvelope::META_DELAY_SECONDS] ?? 0; $delay = is_int($delay) || is_float($delay) ? $delay : 0; if ($delay > 0) { diff --git a/src/QueueProviderInterface.php b/src/QueueProviderInterface.php index c911820..e2914aa 100644 --- a/src/QueueProviderInterface.php +++ b/src/QueueProviderInterface.php @@ -7,7 +7,7 @@ interface QueueProviderInterface { /** - * @param array $meta + * @param array $meta */ public function pushMessage(string $message, array $meta = []): int; From 4c403386615f4e47d5e038be92f818f98f490e8d Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Thu, 9 Jul 2026 11:17:39 +0300 Subject: [PATCH 3/4] Address review feedback --- src/QueueProvider.php | 5 ++--- tests/Support/ExtendedSimpleMessageHandler.php | 7 ++++++- tests/Support/IntegrationTestCase.php | 6 +++++- tests/Unit/Message/MessageTest.php | 4 ++-- 4 files changed, 15 insertions(+), 7 deletions(-) diff --git a/src/QueueProvider.php b/src/QueueProvider.php index 38e37fb..94fe4fa 100644 --- a/src/QueueProvider.php +++ b/src/QueueProvider.php @@ -26,11 +26,10 @@ public function __construct( ) { } - /** - * @throws RedisException - */ /** * @param array $meta + * + * @throws RedisException */ public function pushMessage(string $message, array $meta = []): int { diff --git a/tests/Support/ExtendedSimpleMessageHandler.php b/tests/Support/ExtendedSimpleMessageHandler.php index 1cc87ab..897cc75 100644 --- a/tests/Support/ExtendedSimpleMessageHandler.php +++ b/tests/Support/ExtendedSimpleMessageHandler.php @@ -18,7 +18,12 @@ public function __construct(private FileHelper $fileHelper) public function handle(MessageInterface $message): void { $data = $message->getPayload(); - if (null !== $data) { + if ( + is_array($data) + && is_string($data['file_name'] ?? null) + && is_array($data['payload'] ?? null) + && (is_int($data['payload']['time'] ?? null) || is_string($data['payload']['time'] ?? null)) + ) { $this->fileHelper->put($data['file_name'], $data['payload']['time']); } } diff --git a/tests/Support/IntegrationTestCase.php b/tests/Support/IntegrationTestCase.php index 026b87c..60be450 100644 --- a/tests/Support/IntegrationTestCase.php +++ b/tests/Support/IntegrationTestCase.php @@ -93,7 +93,11 @@ protected function getMessageHandlers(): array 'ext-simple' => [new ExtendedSimpleMessageHandler(new FileHelper()), 'handle'], 'exception-listen' => static function (MessageInterface $message) { $data = $message->getPayload(); - if (null !== $data) { + if ( + is_array($data) + && is_array($data['payload'] ?? null) + && (is_int($data['payload']['time'] ?? null) || is_string($data['payload']['time'] ?? null)) + ) { throw new \RuntimeException((string) $data['payload']['time']); } }, diff --git a/tests/Unit/Message/MessageTest.php b/tests/Unit/Message/MessageTest.php index 451799f..2fc4f7c 100644 --- a/tests/Unit/Message/MessageTest.php +++ b/tests/Unit/Message/MessageTest.php @@ -17,7 +17,7 @@ public function testGetHandlerName(): void $this->assertEquals('handler', $message->getType()); } - public function testGetData(): void + public function testGetPayload(): void { $message = new Message('handler', 'data', []); $this->assertEquals('data', $message->getPayload()); @@ -53,7 +53,7 @@ public function testWithDelay(): void $this->assertEquals(5, $delayedMessage->getMeta()[DelayEnvelope::META_DELAY_SECONDS]); } - public function testFromData(): void + public function testFromPayload(): void { $handlerName = 'test-handler'; $data = ['key' => 'value']; From d0c95ceda20c54db18a48e5751e72b2c80fc30cc Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Thu, 9 Jul 2026 11:26:06 +0300 Subject: [PATCH 4/4] Address self-review findings --- tests/Unit/Message/MessageTest.php | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/tests/Unit/Message/MessageTest.php b/tests/Unit/Message/MessageTest.php index 2fc4f7c..e78be76 100644 --- a/tests/Unit/Message/MessageTest.php +++ b/tests/Unit/Message/MessageTest.php @@ -23,25 +23,25 @@ public function testGetPayload(): void $this->assertEquals('data', $message->getPayload()); } - public function testGetMetadata(): void + public function testGetMeta(): void { - $metadata = ['key' => 'value']; - $message = new Message('handler', 'data', $metadata); - $this->assertEquals($metadata, $message->getMeta()); + $meta = ['key' => 'value']; + $message = new Message('handler', 'data', $meta); + $this->assertEquals($meta, $message->getMeta()); - $message = new Message('handler', 'data', $metadata, 2); - $metadata[DelayEnvelope::META_DELAY_SECONDS] = 2; - $this->assertEquals($metadata, $message->getMeta()); + $message = new Message('handler', 'data', $meta, 2); + $meta[DelayEnvelope::META_DELAY_SECONDS] = 2; + $this->assertEquals($meta, $message->getMeta()); } - public function testWithMetadata(): void + public function testWithMeta(): void { $message = new Message('handler', 'data', []); - $messageWithMetadata = $message->withMeta(['key' => 'value']); + $messageWithMeta = $message->withMeta(['key' => 'value']); - $this->assertNotSame($message, $messageWithMetadata); + $this->assertNotSame($message, $messageWithMeta); $this->assertSame([], $message->getMeta()); - $this->assertSame(['key' => 'value'], $messageWithMetadata->getMeta()); + $this->assertSame(['key' => 'value'], $messageWithMeta->getMeta()); } public function testWithDelay(): void