Cancel the consumer before requeueing a refused delivery, require a true handler result

src/Adapter.php

@@ -110,7 +110,7 @@
             function (AMQPMessage $amqpMessage) use ($handlerCallback, $channel, &$refused): void {
                 if ($refused) {
                     // A message delivered after the handler refused one is not passed to it.
                     $channel->basic_nack($amqpMessage->getDeliveryTag(), false, true);
                     return;
                 }
 
@@ -119,24 +119,25 @@
                 } catch (Throwable $exception) {
                     $consumerTag = $amqpMessage->getConsumerTag();
                     if ($consumerTag !== null) {
                         $channel->basic_cancel($consumerTag);
                     }
 
                     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);
             },
         );
 
@@ -175,7 +176,7 @@
             throw new InvalidArgumentException('Message cannot be delayed to a queue without an exchange. Exchange is mandatory.');
         }
 
         $delayMilliseconds = (int) ceil($delaySeconds * 1000);
 
         return $this->queueProvider
             ->withMessageProperties($this->getDelayMessageProperties($delayMilliseconds))
@@ -194,7 +195,7 @@
      */
     private function getDelayMessageProperties(int $delayMilliseconds): array
     {
         return array_merge(
             $this->queueProvider->getMessageProperties(),
             [
                 'expiration' => (string) $delayMilliseconds,
@@ -208,7 +209,7 @@
         ExchangeSettingsInterface $exchangeSettings,
         int $delayMilliseconds,
     ): QueueSettingsInterface {
         $deliveryTime = time() + (int) ceil($delayMilliseconds / 1000);
 
         return $queueSettings
             ->withName("{$queueSettings->getName()}.dlx.$deliveryTime")

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);

tests/Unit/ExistingMessagesConsumerTest.php

@@ -0,0 +1,87 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Amqp\Tests\Unit;
+
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Message\AMQPMessage;
+use PHPUnit\Framework\TestCase;
+use Yiisoft\Queue\Amqp\ExistingMessagesConsumer;
+use Yiisoft\Queue\Amqp\QueueProviderInterface;
+use Yiisoft\Queue\Amqp\Settings\QueueSettingsInterface;
+use Yiisoft\Queue\Message\GenericMessage;
+use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface;
+
+final class ExistingMessagesConsumerTest extends TestCase
+{
+    /**
+     * @var string[]
+     */
+    private array $calls = [];
+
+    public function testAcknowledgesHandledMessagesUntilQueueIsEmpty(): void
+    {
+        $this->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<list<string>> $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);
+    }
+}

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);
     }
 
     /**