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..94fe4fa 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 @@ -26,15 +27,19 @@ public function __construct( } /** + * @param array $meta + * * @throws RedisException */ - public function pushMessage(string $message, array $metadata = []): int + 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 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) { $this->redis->zadd("$this->channelName.delayed", time() + $delay, $id); } else { diff --git a/src/QueueProviderInterface.php b/src/QueueProviderInterface.php index 4efefd9..e2914aa 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..897cc75 100644 --- a/tests/Support/ExtendedSimpleMessageHandler.php +++ b/tests/Support/ExtendedSimpleMessageHandler.php @@ -17,8 +17,13 @@ public function __construct(private FileHelper $fileHelper) public function handle(MessageInterface $message): void { - $data = $message->getData(); - if (null !== $data) { + $data = $message->getPayload(); + 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 7eb8b4f..60be450 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,8 +92,12 @@ protected function getMessageHandlers(): array return [ 'ext-simple' => [new ExtendedSimpleMessageHandler(new FileHelper()), 'handle'], 'exception-listen' => static function (MessageInterface $message) { - $data = $message->getData(); - if (null !== $data) { + $data = $message->getPayload(); + 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']); } }, @@ -134,7 +139,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..e78be76 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 @@ -16,31 +17,31 @@ 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->getData()); + $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->getMetadata()); + $meta = ['key' => 'value']; + $message = new Message('handler', 'data', $meta); + $this->assertEquals($meta, $message->getMeta()); - $message = new Message('handler', 'data', $metadata, 2); - $metadata['delay'] = 2; - $this->assertEquals($metadata, $message->getMetadata()); + $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->withMetadata(['key' => 'value']); + $messageWithMeta = $message->withMeta(['key' => 'value']); - $this->assertNotSame($message, $messageWithMetadata); - $this->assertSame([], $message->getMetadata()); - $this->assertSame(['key' => 'value'], $messageWithMetadata->getMetadata()); + $this->assertNotSame($message, $messageWithMeta); + $this->assertSame([], $message->getMeta()); + $this->assertSame(['key' => 'value'], $messageWithMeta->getMeta()); } public function testWithDelay(): void @@ -49,19 +50,19 @@ 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 + public function testFromPayload(): 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;