Fix #346: Document and test the loop check before processing a message

docs/guide/en/loops.md

@@ -4,7 +4,8 @@ Yii Queue uses `\Yiisoft\Queue\Cli\LoopInterface` to control long-running execut
 
 The loop is evaluated to determine whether it can continue:
 
-- After each processed message (via `QueueConsumer::run()` / `QueueConsumer::listen()`).
+- Before each message in `QueueConsumer::run()` and `QueueConsumer::listen()`. A message taken from the queue after the
+  loop stopped is not processed and goes back to the queue.
 - On each iteration of `queue:listen-all`.
 
 When `canContinue()` returns `false`, consuming stops gracefully (as soon as the current message is finished).

tests/Unit/QueueTest.php

@@ -4,6 +4,7 @@
 
 namespace Yiisoft\Queue\Tests\Unit;
 
+use Closure;
 use Psr\Log\NullLogger;
 use Yiisoft\Queue\QueueConsumer;
 use Yiisoft\Queue\QueueConsumerInterface;
@@ -119,4 +120,47 @@ public function testMessageHandledBeforeStopIsNotHandledAgain(): void
         self::assertSame(2, $this->executionTimes, 'The message handled before the stop is not handled again.');
         self::assertCount(0, $adapter->getMessagesList());
     }
+
+    public function testConsumerListenReportsMessageProcessedBeforeLoopStops(): void
+    {
+        $adapter = new InMemoryAdapter();
+        $producer = $this->createQueue($adapter);
+        $message = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null)));
+        $consumer = new QueueConsumer($this->getWorker(), $this->createStopAfterFirstExecutionLoop(), new NullLogger(), $adapter);
+
+        $consumer->listen();
+
+        self::assertSame(1, $this->executionTimes);
+        self::assertSame(MessageStatus::DONE, $adapter->status($message->getId()));
+    }
+
+    public function testConsumerListenStopsBeforeProcessingAnotherMessage(): void
+    {
+        $adapter = new InMemoryAdapter();
+        $producer = $this->createQueue($adapter);
+        $first = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null)));
+        $second = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null)));
+        $consumer = new QueueConsumer($this->getWorker(), $this->createStopAfterFirstExecutionLoop(), new NullLogger(), $adapter);
+
+        $consumer->listen();
+
+        self::assertSame(1, $this->executionTimes);
+        self::assertSame(MessageStatus::DONE, $adapter->status($first->getId()));
+        self::assertSame(MessageStatus::WAITING, $adapter->status($second->getId()));
+    }
+
+    private function createStopAfterFirstExecutionLoop(): LoopInterface
+    {
+        return new class (fn(): int => $this->executionTimes) implements LoopInterface {
+            /**
+             * @param Closure(): int $executionTimes
+             */
+            public function __construct(private readonly Closure $executionTimes) {}
+
+            public function canContinue(): bool
+            {
+                return ($this->executionTimes)() === 0;
+            }
+        };
+    }
 }