diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index e20bfb17..c7700b23 100644 --- a/docs/guide/en/loops.md +++ b/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). diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index a0bc23ca..bfe8ba71 100644 --- a/tests/Unit/QueueTest.php +++ b/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; + } + }; + } }