Deleting of expired messages

Author: rosamarsky

Diff for: pkg/dbal/DbalConsumer.php

@@ -78,6 +78,7 @@ public function receiveNoWait(): ?Message
     {
         $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
 
+        $this->removeExpiredMessages();
         $this->redeliverMessages();
 
         // get top message from the queue
@@ -108,6 +109,9 @@ public function reject(Message $message, bool $requeue = false): void
         InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);
 
         if ($requeue) {
+            $message = clone $message;
+            $message->setRedelivered(false);
+
             $this->getContext()->createProducer()->send($this->queue, $message);
 
             return;

Diff for: pkg/dbal/DbalConsumerHelperTrait.php

@@ -92,4 +92,15 @@ protected function redeliverMessages(): void
             ->execute()
         ;
     }
+
+    protected function removeExpiredMessages(): void
+    {
+        $this->getConnection()->createQueryBuilder()
+            ->delete($this->getContext()->getTableName())
+            ->andWhere('(time_to_live IS NOT NULL) AND (time_to_live < :now)')
+            ->setParameter(':now', (int) time(), Type::BIGINT)
+            ->setParameter('redelivered', false, Type::BOOLEAN)
+            ->execute()
+        ;
+    }
 }

Diff for: pkg/dbal/DbalProducer.php

@@ -87,6 +87,7 @@ public function send(Destination $destination, Message $message): void
             'properties' => JSON::encode($message->getProperties()),
             'priority' => $message->getPriority(),
             'queue' => $destination->getQueueName(),
+            'redelivered' => false,
             'delivery_id' => null,
             'redeliver_after' => null,
         ];
@@ -134,6 +135,7 @@ public function send(Destination $destination, Message $message): void
                 'queue' => Type::STRING,
                 'time_to_live' => Type::INTEGER,
                 'delayed_until' => Type::INTEGER,
+                'redelivered' => Type::BOOLEAN,
                 'delivery_id' => Type::STRING,
                 'redeliver_after' => Type::BIGINT,
             ]);

Diff for: pkg/dbal/DbalSubscriptionConsumer.php

@@ -84,6 +84,7 @@ public function consume(int $timeout = 0): void
             }
 
             $now = time();
+            $this->removeExpiredMessages();
             $this->redeliverMessages();
 
             if ($message = $this->fetchMessage($currentQueueNames, $redeliveryDelay)) {

Diff for: pkg/dbal/Tests/DbalConsumerTest.php

@@ -169,7 +169,7 @@ public function testRejectShouldReSendMessageToSameQueueOnRequeue()
         $producerMock
             ->expects($this->once())
             ->method('send')
-            ->with($this->identicalTo($queue), $this->identicalTo($message))
+            ->with($this->identicalTo($queue), $this->isInstanceOf($message))
         ;
 
         $context = $this->createContextMock();

Diff for: pkg/dbal/Tests/Functional/DbalConsumerTest.php

@@ -1,5 +1,7 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\Dbal\Tests\Functional;
 
 use Enqueue\Dbal\DbalContext;
@@ -101,4 +103,52 @@ public function testShouldOrderMessagesWithSamePriorityByPublishedAtDate()
         $consumer->acknowledge($message);
         $this->assertSame($expectedPriority5Body, $message->getBody());
     }
+
+    public function testShouldDeleteExpiredMessage()
+    {
+        $context = $this->context;
+        $queue = $context->createQueue(__METHOD__);
+
+        $consumer = $context->createConsumer($queue);
+
+        // guard
+        $this->assertNull($consumer->receiveNoWait());
+
+        $producer = $context->createProducer();
+
+        $this->context->getDbalConnection()->insert(
+            $this->context->getTableName(), [
+            'id' => 'id',
+            'human_id' => 'id',
+            'published_at' => '123',
+            'body' => 'expiredMessage',
+            'headers' => json_encode([]),
+            'properties' => json_encode([]),
+            'queue' => __METHOD__,
+            'redelivered' => 0,
+            'time_to_live' => time() - 10000,
+        ]);
+
+        $message = $context->createMessage('notExpiredMessage');
+        $message->setRedelivered(false);
+        $producer->send($queue, $message);
+
+        $this->assertSame('2', $this->getQuerySize());
+
+        $message = $consumer->receive(8000);
+
+        $this->assertSame('1', $this->getQuerySize());
+
+        $consumer->acknowledge($message);
+
+        $this->assertSame('0', $this->getQuerySize());
+    }
+
+    private function getQuerySize(): string
+    {
+        return $this->context->getDbalConnection()
+            ->executeQuery('SELECT count(*) FROM '.$this->context->getTableName())
+            ->fetchColumn(0)
+        ;
+    }
 }