diff --git a/src/Adapter/AdapterInterface.php b/src/Adapter/AdapterInterface.php index 69491cb1..e2e5286a 100644 --- a/src/Adapter/AdapterInterface.php +++ b/src/Adapter/AdapterInterface.php @@ -10,9 +10,13 @@ interface AdapterInterface { /** - * Returns the first message from the queue if it exists (null otherwise). + * Passes existing messages to the given handler in FIFO order, until the queue is empty or the handler asks + * to stop. * - * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages + * The handler returns `false` for a message it did not consume: that message must stay in the queue, and no + * further messages are passed. Only a message the handler returned `true` for may be removed from the queue. + * + * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages. */ public function runExisting(callable $handlerCallback): void; @@ -32,6 +36,8 @@ public function push(MessageInterface $message): MessageInterface; /** * Listen to the queue and pass messages to the given handler as they come. * + * The handler returns `false` for a message it did not consume, as in {@see runExisting()}. + * * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages. */ public function subscribe(callable $handlerCallback): void; diff --git a/stubs/InMemoryAdapter.php b/stubs/InMemoryAdapter.php index 5a0ad198..59fe1dd6 100644 --- a/stubs/InMemoryAdapter.php +++ b/stubs/InMemoryAdapter.php @@ -48,9 +48,11 @@ public function push(MessageInterface $message): MessageInterface public function runExisting(callable $handlerCallback): void { - $result = true; - while (isset($this->messages[$this->current]) && $result === true) { - $result = $handlerCallback($this->messages[$this->current]); + while (isset($this->messages[$this->current])) { + if ($handlerCallback($this->messages[$this->current]) !== true) { + break; + } + unset($this->messages[$this->current]); $this->current++; } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index 4923edca..f09568a1 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -12,6 +12,7 @@ use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\MessageStatus; use Yiisoft\Queue\Stubs\InMemoryAdapter; +use Yiisoft\Queue\Stubs\StubLoop; use Yiisoft\Queue\Tests\TestCase; use function count; @@ -81,5 +82,18 @@ public function testConsumerStopsAtLimit(): void $consumer = new QueueConsumer($this->getWorker(), $this->getLoop(), new NullLogger(), $adapter); self::assertSame(1, $consumer->run(1)); self::assertSame(1, $this->executionTimes); + self::assertCount(1, $adapter->getMessagesList(), 'The message that hits the limit stays in the queue.'); + } + + public function testMessageIsNotConsumedWhenLoopCannotContinue(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $producer->push(new GenericMessage('simple', null)); + $producer->push(new GenericMessage('simple', null)); + $consumer = new QueueConsumer($this->getWorker(), new StubLoop(false), new NullLogger(), $adapter); + + self::assertSame(0, $consumer->run()); + self::assertCount(2, $adapter->getMessagesList()); } }