Skip to content

Update for queue message API changes #135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
samdark merged 3 commits into from
Jul 9, 2026
Merged

Update for queue message API changes #135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
29d7ac5
Update for queue message API changes
samdark Jul 9, 2026
1e02fd9
Apply fixes from StyleCI
StyleCIBot Jul 9, 2026
fa8cf61
Address self-review findings
samdark Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/Adapter.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
use Yiisoft\Queue\Cli\LoopInterface;
use Yiisoft\Queue\Message\DelayEnvelope;
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
Expand Down Expand Up @@ -112,7 +112,7 @@
} catch (Throwable $exception) {
$consumerTag = $amqpMessage->getConsumerTag();
if ($consumerTag !== null) {
$channel->basic_cancel($consumerTag);

Check warning on line 115 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "MethodCallRemoval":

@@ @@
             } catch (Throwable $exception) {
                 $consumerTag = $amqpMessage->getConsumerTag();
                 if ($consumerTag !== null) {
-                    $channel->basic_cancel($consumerTag);
+                    
                 }
                 throw $exception;
             }
}

throw $exception;
Expand Down Expand Up @@ -155,7 +155,7 @@
throw new InvalidArgumentException('Message cannot be delayed to a queue without an exchange. Exchange is mandatory.');
}

$delayMilliseconds = (int) ceil($delaySeconds * 1000);

Check warning on line 158 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "RoundingFamily":

@@ @@
         if ($exchangeSettings === null) {
             throw new InvalidArgumentException('Message cannot be delayed to a queue without an exchange. Exchange is mandatory.');
         }
-        $delayMilliseconds = (int) ceil($delaySeconds * 1000);
+        $delayMilliseconds = (int) round($delaySeconds * 1000);
         return $this->queueProvider->withMessageProperties($this->getDelayMessageProperties($delayMilliseconds))->withExchangeSettings($this->getDelayExchangeSettings($exchangeSettings))->withQueueSettings($this->getDelayQueueSettings($this->queueProvider->getQueueSettings(), $exchangeSettings, $delayMilliseconds));
     }
     /**

Check warning on line 158 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "RoundingFamily":

@@ @@
         if ($exchangeSettings === null) {
             throw new InvalidArgumentException('Message cannot be delayed to a queue without an exchange. Exchange is mandatory.');
         }
-        $delayMilliseconds = (int) ceil($delaySeconds * 1000);
+        $delayMilliseconds = (int) floor($delaySeconds * 1000);
         return $this->queueProvider->withMessageProperties($this->getDelayMessageProperties($delayMilliseconds))->withExchangeSettings($this->getDelayExchangeSettings($exchangeSettings))->withQueueSettings($this->getDelayQueueSettings($this->queueProvider->getQueueSettings(), $exchangeSettings, $delayMilliseconds));
     }
     /**

return $this->queueProvider
->withMessageProperties($this->getDelayMessageProperties($delayMilliseconds))
Expand All @@ -174,7 +174,7 @@
*/
private function getDelayMessageProperties(int $delayMilliseconds): array
{
return array_merge(

Check warning on line 177 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "UnwrapArrayMerge":

@@ @@
      */
     private function getDelayMessageProperties(int $delayMilliseconds): array
     {
-        return array_merge($this->queueProvider->getMessageProperties(), ['expiration' => (string) $delayMilliseconds, 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
+        return ['expiration' => (string) $delayMilliseconds, 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT];
     }
     private function getDelayQueueSettings(QueueSettingsInterface $queueSettings, ExchangeSettingsInterface $exchangeSettings, int $delayMilliseconds): QueueSettingsInterface
     {
$this->queueProvider->getMessageProperties(),
[
'expiration' => (string) $delayMilliseconds,
Expand All @@ -188,7 +188,7 @@
ExchangeSettingsInterface $exchangeSettings,
int $delayMilliseconds,
): QueueSettingsInterface {
$deliveryTime = time() + (int) ceil($delayMilliseconds / 1000);

Check warning on line 191 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "Division":

@@ @@
     }
     private function getDelayQueueSettings(QueueSettingsInterface $queueSettings, ExchangeSettingsInterface $exchangeSettings, int $delayMilliseconds): QueueSettingsInterface
     {
-        $deliveryTime = time() + (int) ceil($delayMilliseconds / 1000);
+        $deliveryTime = time() + (int) ceil($delayMilliseconds * 1000);
         return $queueSettings->withName("{$queueSettings->getName()}.dlx.{$deliveryTime}")->withAutoDeletable(true)->withArguments(['x-dead-letter-exchange' => ['S', $exchangeSettings->getName()], 'x-expires' => ['I', $delayMilliseconds + 30000], 'x-message-ttl' => ['I', $delayMilliseconds]]);
     }
     private function getDelayExchangeSettings(ExchangeSettingsInterface $exchangeSettings): ExchangeSettingsInterface

Check warning on line 191 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "IncrementInteger":

@@ @@
     }
     private function getDelayQueueSettings(QueueSettingsInterface $queueSettings, ExchangeSettingsInterface $exchangeSettings, int $delayMilliseconds): QueueSettingsInterface
     {
-        $deliveryTime = time() + (int) ceil($delayMilliseconds / 1000);
+        $deliveryTime = time() + (int) ceil($delayMilliseconds / 1001);
         return $queueSettings->withName("{$queueSettings->getName()}.dlx.{$deliveryTime}")->withAutoDeletable(true)->withArguments(['x-dead-letter-exchange' => ['S', $exchangeSettings->getName()], 'x-expires' => ['I', $delayMilliseconds + 30000], 'x-message-ttl' => ['I', $delayMilliseconds]]);
     }
     private function getDelayExchangeSettings(ExchangeSettingsInterface $exchangeSettings): ExchangeSettingsInterface

Check warning on line 191 in src/Adapter.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "DecrementInteger":

@@ @@
     }
     private function getDelayQueueSettings(QueueSettingsInterface $queueSettings, ExchangeSettingsInterface $exchangeSettings, int $delayMilliseconds): QueueSettingsInterface
     {
-        $deliveryTime = time() + (int) ceil($delayMilliseconds / 1000);
+        $deliveryTime = time() + (int) ceil($delayMilliseconds / 999);
         return $queueSettings->withName("{$queueSettings->getName()}.dlx.{$deliveryTime}")->withAutoDeletable(true)->withArguments(['x-dead-letter-exchange' => ['S', $exchangeSettings->getName()], 'x-expires' => ['I', $delayMilliseconds + 30000], 'x-message-ttl' => ['I', $delayMilliseconds]]);
     }
     private function getDelayExchangeSettings(ExchangeSettingsInterface $exchangeSettings): ExchangeSettingsInterface

return $queueSettings
->withName("{$queueSettings->getName()}.dlx.$deliveryTime")
Expand Down
2 changes: 1 addition & 1 deletion src/ExistingMessagesConsumer.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
use PhpAmqpLib\Message\AMQPMessage;
use Throwable;
use Yiisoft\Queue\Message\MessageInterface;
use Yiisoft\Queue\Message\MessageSerializerInterface;
use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface;

/**
* @internal
Expand All @@ -33,7 +33,7 @@
$channel->basic_consume(
$this->queueProvider->getQueueSettings()->getName(),
$consumerTag,
false,

Check warning on line 36 in src/ExistingMessagesConsumer.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "FalseValue":

@@ @@
         $channel = $this->queueProvider->getChannel();
         $consumerTag = uniqid(more_entropy: true);
         try {
-            $channel->basic_consume($this->queueProvider->getQueueSettings()->getName(), $consumerTag, false, false, false, false, function (AMQPMessage $amqpMessage) use ($callback, $channel): void {
+            $channel->basic_consume($this->queueProvider->getQueueSettings()->getName(), $consumerTag, true, false, false, false, function (AMQPMessage $amqpMessage) use ($callback, $channel): void {
                 try {
                     $message = $this->serializer->unserialize($amqpMessage->getBody());
                     if ($this->messageConsumed = $callback($message)) {
false,
false,
false,
Expand All @@ -59,7 +59,7 @@
$channel->wait(null, true);
} while ($this->messageConsumed === true);
} finally {
$channel->basic_cancel($consumerTag);

Check warning on line 62 in src/ExistingMessagesConsumer.php

View workflow job for this annotation

GitHub Actions / PHP 8.4-ubuntu-latest 

Escaped Mutant for Mutator "MethodCallRemoval":

@@ @@
                 $channel->wait(null, true);
             } while ($this->messageConsumed === true);
         } finally {
-            $channel->basic_cancel($consumerTag);
+            
             $this->queueProvider->channelClose();
         }
     }
 }
$this->queueProvider->channelClose();
}
}
Expand Down
7 changes: 4 additions & 3 deletions tests/Benchmark/QueueConsumeBench.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,8 @@
use Yiisoft\Queue\Amqp\QueueProvider;
use Yiisoft\Queue\Amqp\Settings\Queue as QueueSettings;
use Yiisoft\Queue\Cli\SimpleLoop;
use Yiisoft\Queue\Message\JsonMessageSerializer;
use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
use Yiisoft\Queue\Message\Serializer\MessageSerializer;
use Yiisoft\Queue\Amqp\Tests\Support\TestMessage as Message;

#[BeforeClassMethods('cleanupQueue')]
Expand Down Expand Up @@ -42,7 +43,7 @@ public function benchConsume(): void
public function pushMessagesForConsume(): void
{
for ($i = 0; $i < self::MESSAGE_COUNT; $i++) {
$this->adapter->push(Message::fromData('test', ['payload' => 'test']));
$this->adapter->push(Message::fromPayload('test', ['payload' => 'test']));
}
}

Expand All @@ -54,7 +55,7 @@ public static function cleanupQueue(): void
private static function getAdapter(): Adapter
{
$loop = new SimpleLoop();
$serializer = new JsonMessageSerializer();
$serializer = new MessageSerializer(new JsonMessageEncoder());
$queueProvider = new QueueProvider(
new AMQPStreamConnection(
getenv('RABBITMQ_HOST'),
Expand Down
9 changes: 5 additions & 4 deletions tests/Benchmark/QueuePushBench.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@
use Yiisoft\Queue\Amqp\QueueProvider;
use Yiisoft\Queue\Amqp\Settings\Queue as QueueSettings;
use Yiisoft\Queue\Cli\SimpleLoop;
use Yiisoft\Queue\Message\JsonMessageSerializer;
use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
use Yiisoft\Queue\Message\Serializer\MessageSerializer;
use Yiisoft\Queue\Amqp\Tests\Support\TestMessage as Message;

final class QueuePushBench
Expand All @@ -27,7 +28,7 @@ final class QueuePushBench
public function __construct()
{
$loop = new SimpleLoop();
$serializer = new JsonMessageSerializer();
$serializer = new MessageSerializer(new JsonMessageEncoder());
$queueProvider = new QueueProvider(
new AMQPStreamConnection(
getenv('RABBITMQ_HOST'),
Expand All @@ -53,7 +54,7 @@ public function __construct()
#[OutputTimeUnit('seconds')]
public function benchPush(): void
{
$this->adapter->push(Message::fromData('test', ['payload' => 'test']));
$this->adapter->push(Message::fromPayload('test', ['payload' => 'test']));
}

/**
Expand All @@ -67,7 +68,7 @@ public function benchPush(): void
#[OutputTimeUnit('seconds')]
public function benchPushBatch(): void
{
$message = Message::fromData('test', ['payload' => 'test']);
$message = Message::fromPayload('test', ['payload' => 'test']);
for ($i = 0; $i < 100; $i++) {
$this->adapter->push($message);
}
Expand Down
11 changes: 6 additions & 5 deletions tests/Integration/ConsumeExistingMessagesTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@
use Yiisoft\Queue\Amqp\QueueProvider;
use Yiisoft\Queue\Amqp\Settings\Queue as QueueSettings;
use Yiisoft\Queue\Cli\SimpleLoop;
use Yiisoft\Queue\Message\JsonMessageSerializer;
use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
use Yiisoft\Queue\Message\Serializer\MessageSerializer;
use Yiisoft\Queue\Amqp\Tests\Support\TestMessage as Message;
use PhpAmqpLib\Connection\AMQPStreamConnection;

Expand All @@ -17,7 +18,7 @@ class ConsumeExistingMessagesTest extends TestCase
public function testConsumeExistingMessages(): void
{
$loop = new SimpleLoop();
$serializer = new JsonMessageSerializer();
$serializer = new MessageSerializer(new JsonMessageEncoder());
$queueProvider = new QueueProvider(
new AMQPStreamConnection(
getenv('RABBITMQ_HOST'),
Expand All @@ -31,7 +32,7 @@ public function testConsumeExistingMessages(): void

$messageCount = 10;
for ($i = 0; $i < $messageCount; $i++) {
$adapter->push(Message::fromData('test', ['payload' => 'test']));
$adapter->push(Message::fromPayload('test', ['payload' => 'test']));
}

// wait for messages to be ready to consume
Expand All @@ -49,7 +50,7 @@ public function testConsumeExistingMessages(): void
public function testConsumeExistingMessagesByOne(): void
{
$loop = new SimpleLoop();
$serializer = new JsonMessageSerializer();
$serializer = new MessageSerializer(new JsonMessageEncoder());
$queueProvider = new QueueProvider(
new AMQPStreamConnection(
getenv('RABBITMQ_HOST'),
Expand All @@ -63,7 +64,7 @@ public function testConsumeExistingMessagesByOne(): void

$messageCount = 10;
for ($i = 0; $i < $messageCount; $i++) {
$adapter->push(Message::fromData('test', ['payload' => 'test']));
$adapter->push(Message::fromPayload('test', ['payload' => 'test']));
}

// wait for messages to be ready to consume
Expand Down
20 changes: 12 additions & 8 deletions tests/Integration/DelayMiddlewareTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,13 @@
use Yiisoft\Queue\Amqp\Tests\Support\FileHelper;
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\Amqp\Tests\Support\TestMessage as Message;
use Yiisoft\Queue\Middleware\CallableFactory;
use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
use Yiisoft\Queue\Provider\QueueProviderInterface;
use Yiisoft\Queue\Queue;
use Yiisoft\Queue\Worker\WorkerInterface;
use Yiisoft\Test\Support\Container\SimpleContainer;
Expand All @@ -44,14 +46,14 @@ public function testMainFlow(): void
new QueueSettings(),
new ExchangeSettings('yii-queue'),
),
new JsonMessageSerializer(),
new MessageSerializer(new JsonMessageEncoder()),
new SignalLoop(),
);
$queue = $this->makeQueue($adapter)->withMiddlewaresAdded(new DelayMiddleware(3));
$queue = $this->makeQueue($adapter, new DelayMiddleware(3));

$time = time();
$queue->push(
Message::fromData('simple', 'test-delay-middleware-main'),
Message::fromPayload('simple', 'test-delay-middleware-main'),
);

sleep(2);
Expand All @@ -72,19 +74,19 @@ public function testMainFlowWithFakeAdapter(): void
new QueueSettings(),
new ExchangeSettings('yii-queue'),
),
new JsonMessageSerializer(),
new MessageSerializer(new JsonMessageEncoder()),
new SignalLoop(),
);
$queue = $this->makeQueue($adapter)->withMiddlewaresAdded(new DelayMiddleware(3));
$queue = $this->makeQueue($adapter, new DelayMiddleware(3));

$this->expectException(\LogicException::class);
$this->expectExceptionMessage('Method not implemented');
$queue->push(
Message::fromData('simple', 'test-delay-middleware-main'),
Message::fromPayload('simple', 'test-delay-middleware-main'),
);
}

private function makeQueue(AdapterInterface $adapter): Queue
private function makeQueue(AdapterInterface $adapter, mixed ...$middlewareDefinitions): Queue
{
return new Queue(
$this->createMock(WorkerInterface::class),
Expand All @@ -97,6 +99,8 @@ private function makeQueue(AdapterInterface $adapter): Queue
),
),
$adapter,
QueueProviderInterface::DEFAULT_QUEUE,
...$middlewareDefinitions,
);
}
}
9 changes: 7 additions & 2 deletions tests/Support/ExtendedSimpleMessageHandler.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,13 @@ public function __construct(private readonly 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']);
}
}
Expand Down
2 changes: 1 addition & 1 deletion tests/Support/FakeAdapter.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
use Yiisoft\Queue\Amqp\QueueProviderInterface;
use Yiisoft\Queue\Cli\LoopInterface;
use Yiisoft\Queue\Message\MessageInterface;
use Yiisoft\Queue\Message\MessageSerializerInterface;
use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface;
use Yiisoft\Queue\MessageStatus;

final class FakeAdapter implements AdapterInterface
Expand Down
5 changes: 4 additions & 1 deletion tests/Support/SimpleMessageHandler.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@ public function __construct(private readonly FileHelper $fileHelper)

public function __invoke(MessageInterface $message): void
{
$this->fileHelper->put($message->getData(), time());
$fileName = $message->getPayload();
if (is_string($fileName)) {
$this->fileHelper->put($fileName, time());
}
}
}
24 changes: 14 additions & 10 deletions tests/Support/TestMessage.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,35 +8,39 @@

final class TestMessage implements MessageInterface
{
/**
* @param array|bool|float|int|string|null $payload
* @param array<string, array|bool|float|int|string|null> $meta
*/
public function __construct(
private readonly string $type,
private readonly mixed $data,
private readonly array $metadata = [],
private readonly bool|int|float|string|array|null $payload,
private readonly array $meta = [],
) {
}

public static function fromData(string $type, mixed $data, array $metadata = []): self
public static function fromPayload(string $type, bool|int|float|string|array|null $payload): static
{
return new self($type, $data, $metadata);
return new self($type, $payload);
}

public function getType(): string
{
return $this->type;
}

public function getData(): mixed
public function getPayload(): bool|int|float|string|array|null
{
return $this->data;
return $this->payload;
}

public function getMetadata(): array
public function getMeta(): array
{
return $this->metadata;
return $this->meta;
}

public function withMetadata(array $metadata): static
public function withMeta(array $meta): static
{
return new self($this->type, $this->data, $metadata);
return new self($this->type, $this->payload, $meta);
}
}
4 changes: 2 additions & 2 deletions tests/Unit/DelayMiddlewareTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ public function testImmutable(): void

public function testProcessPushAddsDelayEnvelope(): void
{
$message = Message::fromData('simple', null);
$message = Message::fromPayload('simple', null);
$handler = new class () implements PushHandlerInterface {
public function handlePush(MessageInterface $message): MessageInterface
{
Expand All @@ -47,7 +47,7 @@ public function handlePush(MessageInterface $message): MessageInterface

public function testProcessPushSkipsNonPositiveDelay(): void
{
$message = Message::fromData('simple', null);
$message = Message::fromPayload('simple', null);
$handler = new class () implements PushHandlerInterface {
public function handlePush(MessageInterface $message): MessageInterface
{
Expand Down
11 changes: 6 additions & 5 deletions tests/Unit/QueueProviderTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@
use Yiisoft\Queue\Amqp\Settings\Queue as QueueSettings;
use Yiisoft\Queue\Amqp\Settings\QueueSettingsInterface;
use Yiisoft\Queue\Amqp\Tests\Support\FileHelper;
use Yiisoft\Queue\Message\JsonMessageSerializer;
use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
use Yiisoft\Queue\Message\Serializer\MessageSerializer;
use Yiisoft\Queue\Amqp\Tests\Support\TestMessage as Message;

final class QueueProviderTest extends UnitTestCase
Expand All @@ -34,7 +35,7 @@ public function testWithQueueAndExchangeSettings(): void
->withExchangeSettings(
new ExchangeSettings($this->exchangeName)
),
new JsonMessageSerializer(),
new MessageSerializer(new JsonMessageEncoder()),
$this->getLoop(),
);

Expand All @@ -43,7 +44,7 @@ public function testWithQueueAndExchangeSettings(): void
$fileHelper = new FileHelper();
$time = time();
$queue->push(
Message::fromData('ext-simple', ['file_name' => 'test-with-queue-settings', 'payload' => ['time' => $time]])
Message::fromPayload('ext-simple', ['file_name' => 'test-with-queue-settings', 'payload' => ['time' => $time]])
);

$message = $this
Expand All @@ -61,7 +62,7 @@ public function testWithQueueAndExchangeSettings(): void
self::assertEquals($time, $result);

$messageBody = json_decode($message->getBody(), true, 512, JSON_THROW_ON_ERROR);
self::assertEquals($messageBody['data']['payload']['time'], $result);
self::assertEquals($messageBody['payload']['payload']['time'], $result);
}

public function testWithQueueNameExchangeDeclaredException(): void
Expand All @@ -81,7 +82,7 @@ public function testWithQueueNameExchangeDeclaredException(): void
new ExchangeSettings('yii-queue-test-with-queue-name')
)
->withQueueName('yii-queue-test-queue-name'),
new JsonMessageSerializer(),
new MessageSerializer(new JsonMessageEncoder()),
$this->getLoop(),
);
}
Expand Down
7 changes: 4 additions & 3 deletions tests/Unit/QueueSettingsTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@
use Yiisoft\Queue\Amqp\Settings\Exchange as ExchangeSettings;
use Yiisoft\Queue\Amqp\Settings\QosSettings;
use Yiisoft\Queue\Amqp\Settings\Queue as QueueSettings;
use Yiisoft\Queue\Message\JsonMessageSerializer;
use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
use Yiisoft\Queue\Message\Serializer\MessageSerializer;
use Yiisoft\Queue\Amqp\Tests\Support\TestMessage as Message;

final class QueueSettingsTest extends UnitTestCase
Expand Down Expand Up @@ -97,13 +98,13 @@ public function testArgumentsXExpires(): void
->withExchangeSettings(
new ExchangeSettings('yii-queue-test-queue-settings-arg')
),
new JsonMessageSerializer(),
new MessageSerializer(new JsonMessageEncoder()),
$this->getLoop(),
);

$this->getQueueWithAdapter($adapter)
->push(
Message::fromData('ext-simple', ['payload' => time()])
Message::fromPayload('ext-simple', ['payload' => time()])
);

sleep(2);
Expand Down
Loading
Loading