diff --git a/src/Internal/functions.php b/src/Internal/functions.php index e604eb3..d61989d 100644 --- a/src/Internal/functions.php +++ b/src/Internal/functions.php @@ -42,6 +42,7 @@ function consume( $watcher, $completionMarker, $context, + $polls, ): void { $watcher->watch(); @@ -66,6 +67,12 @@ function consume( } $tx->commit(); + + // A full batch means the queue may hold more messages. Inserts that already happened will not notify + // again, so waiting for the next signal would leave the backlog idle for up to $pollInterval. + if (\count($messages) === $config->batch && !$polls->isComplete()) { + $polls->pushAsync(null)->ignore(); + } } catch (\Throwable $e) { $tx->rollback(); $completionMarker->error($e); diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php index ce2f1f0..49f95f5 100644 --- a/tests/PgmqTest.php +++ b/tests/PgmqTest.php @@ -8,6 +8,7 @@ use Amp\Postgres\PostgresConnection; use Amp\Postgres\PostgresConnectionPool; use Amp\Postgres\PostgresQueryError; +use Amp\TimeoutCancellation; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\Attributes\DoesNotPerformAssertions; use PHPUnit\Framework\TestCase; @@ -326,6 +327,38 @@ static function (array $messages, ConsumeController $ctrl) use (&$consumed): voi self::assertSame(0, $queue->metrics()->length); } + public function testConsumerDrainsBacklogWithoutWaitingForPollInterval(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + $queue->sendBatch(array_fill(0, 25, new SendMessage(self::TESTING_MESSAGE))); + + $consumed = 0; + + $consumer = createConsumer($this->pg); + $context = $consumer->consume( + static function (array $messages, ConsumeController $ctrl) use (&$consumed): void { + $consumed += \count($messages); + $ctrl->ack($messages); + + if ($consumed === 25) { + $ctrl->stop(); + } + }, + new ConsumeConfig( + queue: $queue->name, + batch: 10, + pollInterval: TimeSpan::fromSeconds(10), + ), + ); + + // The backlog is already there: no further inserts will wake the consumer up, so a full batch must + // trigger the next read immediately instead of waiting for the poll interval (2 x 10 s for 25 messages). + $context->awaitCompletion(new TimeoutCancellation(2)); + + self::assertSame(25, $consumed); + self::assertSame(0, $queue->metrics()->length); + } + public function testNackBatch(): void { $queue = createQueue($this->pg, $this->randomQueueName());