From 27aaca34ca379106603beb417b4d457bf9de2413 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Fri, 30 Oct 2026 01:05:20 +0300 Subject: [PATCH] Cancel the consumer before requeueing a refused delivery, require a `true` handler result - `Adapter::subscribe()` cancels the consumer before it nacks the refused delivery, so the broker doesn't deliver the requeued message to the consumer that is stopping. - A delivery counts as handled only when the handler returns `true`, in both `Adapter::subscribe()` and `ExistingMessagesConsumer`. --- src/Adapter.php | 7 +- src/ExistingMessagesConsumer.php | 2 +- tests/Unit/ExistingMessagesConsumerTest.php | 87 +++++++++++++++++++++ tests/Unit/SubscribeTest.php | 13 ++- 4 files changed, 103 insertions(+), 6 deletions(-) create mode 100644 tests/Unit/ExistingMessagesConsumerTest.php diff --git a/src/Adapter.php b/src/Adapter.php index 7b25ffd..6290b3a 100644 --- a/src/Adapter.php +++ b/src/Adapter.php @@ -125,18 +125,19 @@ function (AMQPMessage $amqpMessage) use ($handlerCallback, $channel, &$refused): throw $exception; } - if ($handled) { + if ($handled === true) { $channel->basic_ack($amqpMessage->getDeliveryTag()); return; } - // The handler did not process the message and cannot continue: return it to the queue and stop. + // The handler did not process the message and cannot continue: stop and return it to the queue. + // The consumer is cancelled first, so that the broker doesn't deliver the message to it again. $refused = true; - $channel->basic_nack($amqpMessage->getDeliveryTag(), false, true); $consumerTag = $amqpMessage->getConsumerTag(); if ($consumerTag !== null) { $channel->basic_cancel($consumerTag); } + $channel->basic_nack($amqpMessage->getDeliveryTag(), false, true); }, ); diff --git a/src/ExistingMessagesConsumer.php b/src/ExistingMessagesConsumer.php index 0a8d214..13ae731 100644 --- a/src/ExistingMessagesConsumer.php +++ b/src/ExistingMessagesConsumer.php @@ -39,7 +39,7 @@ public function consume(callable $callback): void function (AMQPMessage $amqpMessage) use ($callback, $channel): void { try { $message = $this->serializer->unserialize($amqpMessage->getBody()); - if ($this->messageConsumed = $callback($message)) { + if ($this->messageConsumed = $callback($message) === true) { $channel->basic_ack($amqpMessage->getDeliveryTag()); } else { $channel->basic_nack($amqpMessage->getDeliveryTag(), false, true); diff --git a/tests/Unit/ExistingMessagesConsumerTest.php b/tests/Unit/ExistingMessagesConsumerTest.php new file mode 100644 index 0000000..3e774aa --- /dev/null +++ b/tests/Unit/ExistingMessagesConsumerTest.php @@ -0,0 +1,87 @@ +createConsumer([['1'], ['2'], []])->consume(static fn(): bool => true); + + self::assertSame(['ack 1', 'ack 2', 'cancel', 'close'], $this->calls); + } + + public function testRequeuesMessageWhenHandlerReturnsNonBoolean(): void + { + $this->createConsumer([['1'], ['2']])->consume(static fn(): int => 1); + + self::assertSame(['nack 1 requeue', 'cancel', 'close'], $this->calls); + } + + /** + * @param list> $deliveries Messages delivered by each `wait()` call. + */ + private function createConsumer(array $deliveries): ExistingMessagesConsumer + { + $consumer = null; + $channel = $this->createMock(AMQPChannel::class); + $channel->method('basic_consume')->willReturnCallback( + static function (mixed ...$arguments) use (&$consumer): string { + $consumer = $arguments[6]; + return (string) $arguments[1]; + }, + ); + $channel->method('wait')->willReturnCallback( + static function () use (&$consumer, &$deliveries): void { + foreach (array_shift($deliveries) ?? [] as $body) { + $message = new AMQPMessage($body); + $message->setDeliveryInfo((int) $body, false, '', 'queue'); + $consumer($message); + } + }, + ); + $channel->method('basic_ack')->willReturnCallback(function (int $tag): void { + $this->calls[] = "ack $tag"; + }); + $channel->method('basic_nack')->willReturnCallback(function (int $tag, bool $multiple, bool $requeue): void { + $this->calls[] = "nack $tag" . ($requeue ? ' requeue' : ''); + }); + $channel->method('basic_cancel')->willReturnCallback(function (): mixed { + $this->calls[] = 'cancel'; + return null; + }); + + $settings = $this->createMock(QueueSettingsInterface::class); + $settings->method('getName')->willReturn('queue'); + $provider = $this->createMock(QueueProviderInterface::class); + $provider->method('getChannel')->willReturn($channel); + $provider->method('getQueueSettings')->willReturn($settings); + $provider->method('channelClose')->willReturnCallback(function (): void { + $this->calls[] = 'close'; + }); + + $serializer = $this->createMock(MessageSerializerInterface::class); + $serializer->method('unserialize')->willReturnCallback( + static fn(string $body): MessageInterface => new GenericMessage('job', $body), + ); + + return new ExistingMessagesConsumer($provider, $serializer); + } +} diff --git a/tests/Unit/SubscribeTest.php b/tests/Unit/SubscribeTest.php index dbf52dd..0570cac 100644 --- a/tests/Unit/SubscribeTest.php +++ b/tests/Unit/SubscribeTest.php @@ -44,7 +44,7 @@ static function (MessageInterface $message) use (&$handled): bool { ); self::assertSame(['1'], $handled); - self::assertSame(['nack 1 requeue', 'cancel consumer'], $this->calls); + self::assertSame(['cancel consumer', 'nack 1 requeue'], $this->calls); } public function testRequeuesMessagesDeliveredAfterRefusalWithoutHandlingThem(): void @@ -60,7 +60,16 @@ static function (MessageInterface $message) use (&$handled): bool { ); self::assertSame(['1', '2'], $handled); - self::assertSame(['ack 1', 'nack 2 requeue', 'cancel consumer', 'nack 3 requeue'], $this->calls); + self::assertSame(['ack 1', 'cancel consumer', 'nack 2 requeue', 'nack 3 requeue'], $this->calls); + } + + public function testRequeuesMessageWhenHandlerReturnsNonBoolean(): void + { + $channel = $this->createChannel([['1']]); + + $this->createAdapter($channel, [true, true])->subscribe(static fn(): int => 1); + + self::assertSame(['cancel consumer', 'nack 1 requeue'], $this->calls); } /**