[Messenger] Add keepalive support

Author: silasjoisten

src/Symfony/Component/Messenger/Bridge/Doctrine/CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.3
+---
+
+ * Add "keepalive" support
+
 7.1
 ---
 

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/ConnectionTest.php

@@ -299,6 +299,97 @@ public function testSendLastInsertIdReturnsInteger()
         self::assertSame('1', $id);
     }
 
+    public function testKeepalive()
+    {
+        $queryBuilder = $this->getQueryBuilderMock();
+        $driverConnection = $this->getDBALConnectionMock();
+
+        $connection = new Connection(['redeliver_timeout' => 30, 'table_name' => 'messenger_messages'], $driverConnection);
+
+        $queryBuilder->expects($this->once())
+            ->method('update')
+            ->with('messenger_messages')
+            ->willReturnSelf();
+
+        $queryBuilder->expects($this->once())
+            ->method('set')
+            ->with('delivered_at', '?')
+            ->willReturnSelf();
+
+        $queryBuilder->expects($this->once())
+            ->method('where')
+            ->with('id = ?')
+            ->willReturnSelf();
+
+        $driverConnection->expects($this->once())
+            ->method('beginTransaction');
+
+        $driverConnection->expects($this->once())
+            ->method('createQueryBuilder')
+            ->willReturn($queryBuilder);
+
+        $driverConnection->expects($this->once())
+            ->method('commit');
+
+        $connection->keepalive('1');
+    }
+
+    public function testKeepaliveRollback()
+    {
+        $queryBuilder = $this->getQueryBuilderMock();
+        $driverConnection = $this->getDBALConnectionMock();
+
+        $connection = new Connection(['redeliver_timeout' => 30, 'table_name' => 'messenger_messages'], $driverConnection);
+
+        $queryBuilder->expects($this->once())
+            ->method('update')
+            ->with('messenger_messages')
+            ->willReturnSelf();
+
+        $queryBuilder->expects($this->once())
+            ->method('set')
+            ->with('delivered_at', '?')
+            ->willReturnSelf();
+
+        $queryBuilder->expects($this->once())
+            ->method('where')
+            ->with('id = ?')
+            ->willReturnSelf();
+
+        $driverConnection->expects($this->once())
+            ->method('beginTransaction');
+
+        $driverConnection->expects($this->once())
+            ->method('createQueryBuilder')
+            ->willReturn($queryBuilder);
+
+        $driverConnection->expects($this->once())
+            ->method('executeStatement')
+            ->willThrowException($this->createMock(DBALException::class));
+
+        $driverConnection->expects($this->never())
+            ->method('commit');
+
+        $driverConnection->expects($this->once())
+            ->method('rollBack');
+
+        $this->expectException(TransportException::class);
+
+        $connection->keepalive('1');
+    }
+
+    public function testKeepaliveThrowsExceptionWhenRedeliverTimeoutIsLessThenInterval()
+    {
+        $driverConnection = $this->getDBALConnectionMock();
+
+        $connection = new Connection(['redeliver_timeout' => 30], $driverConnection);
+
+        $this->expectException(TransportException::class);
+        $this->expectExceptionMessage('Doctrine redeliver_timeout (30s) cannot be smaller than the keepalive interval (60s).');
+
+        $connection->keepalive('1', 60);
+    }
+
     private function getDBALConnectionMock()
     {
         $driverConnection = $this->createMock(DBALConnection::class);

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/DoctrineReceiverTest.php

@@ -353,6 +353,22 @@ public function testRejectThrowsException()
         $receiver->reject($envelope);
     }
 
+    public function testKeepalive()
+    {
+        $serializer = $this->createSerializer();
+        $connection = $this->createMock(Connection::class);
+
+        $envelope = new Envelope(new \stdClass(), [new DoctrineReceivedStamp('1')]);
+        $receiver = new DoctrineReceiver($connection, $serializer);
+
+        $connection
+            ->expects($this->once())
+            ->method('keepalive')
+            ->with('1');
+
+        $receiver->keepalive($envelope);
+    }
+
     private function createDoctrineEnvelope(): array
     {
         return [

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/DoctrineTransportTest.php

@@ -16,6 +16,7 @@
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Bridge\Doctrine\Tests\Fixtures\DummyMessage;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;
+use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineReceivedStamp;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -69,6 +70,22 @@ public function testConfigureSchema()
         $transport->configureSchema($schema, $dbalConnection, static fn () => true);
     }
 
+    public function testKeepalive()
+    {
+        $transport = $this->getTransport(
+            null,
+            $connection = $this->createMock(Connection::class)
+        );
+
+        $envelope = new Envelope(new \stdClass(), [new DoctrineReceivedStamp('1')]);
+
+        $connection->expects($this->once())
+            ->method('keepalive')
+            ->with('1');
+
+        $transport->keepalive($envelope);
+    }
+
     private function getTransport(?SerializerInterface $serializer = null, ?Connection $connection = null): DoctrineTransport
     {
         $serializer ??= $this->createMock(SerializerInterface::class);

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/Connection.php

@@ -286,6 +286,34 @@ public function reject(string $id): bool
         }
     }
 
+    public function keepalive(string $id, ?int $seconds = null): void
+    {
+        // Check if the redeliver timeout is smaller than the keepalive interval
+        if (null !== $seconds && $this->configuration['redeliver_timeout'] < $seconds) {
+            throw new TransportException(\sprintf('Doctrine redeliver_timeout (%ds) cannot be smaller than the keepalive interval (%ds).', $this->configuration['redeliver_timeout'], $seconds));
+        }
+
+        $this->driverConnection->beginTransaction();
+        try {
+            $queryBuilder = $this->driverConnection->createQueryBuilder()
+                ->update($this->configuration['table_name'])
+                ->set('delivered_at', '?')
+                ->where('id = ?');
+            $now = new \DateTimeImmutable('UTC');
+            $this->executeStatement($queryBuilder->getSQL(), [
+                $now,
+                $id,
+            ], [
+                Types::DATETIME_IMMUTABLE,
+            ]);
+
+            $this->driverConnection->commit();
+        } catch (\Throwable $e) {
+            $this->driverConnection->rollBack();
+            throw new TransportException($e->getMessage(), 0, $e);
+        }
+    }
+
     public function setup(): void
     {
         $configuration = $this->driverConnection->getConfiguration();

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineReceiver.php

@@ -18,6 +18,7 @@
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
+use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
@@ -26,7 +27,7 @@
 /**
  * @author Vincent Touzet <[email protected]>
  */
-class DoctrineReceiver implements ListableReceiverInterface, MessageCountAwareInterface
+class DoctrineReceiver implements ListableReceiverInterface, MessageCountAwareInterface, KeepaliveReceiverInterface
 {
     private const MAX_RETRIES = 3;
     private int $retryingSafetyCounter = 0;
@@ -72,6 +73,11 @@ public function ack(Envelope $envelope): void
         });
     }
 
+    public function keepalive(Envelope $envelope, ?int $seconds = null): void
+    {
+        $this->connection->keepalive($this->findDoctrineReceivedStamp($envelope)->getId(), $seconds);
+    }
+
     public function reject(Envelope $envelope): void
     {
         $this->withRetryableExceptionRetry(function () use ($envelope) {

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineTransport.php

@@ -15,6 +15,7 @@
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Schema\Table;
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -24,7 +25,7 @@
 /**
  * @author Vincent Touzet <[email protected]>
  */
-class DoctrineTransport implements TransportInterface, SetupableTransportInterface, MessageCountAwareInterface, ListableReceiverInterface
+class DoctrineTransport implements TransportInterface, SetupableTransportInterface, MessageCountAwareInterface, ListableReceiverInterface, KeepaliveReceiverInterface
 {
     private DoctrineReceiver $receiver;
     private DoctrineSender $sender;
@@ -50,6 +51,11 @@ public function reject(Envelope $envelope): void
         $this->getReceiver()->reject($envelope);
     }
 
+    public function keepalive(Envelope $envelope, ?int $seconds = null): void
+    {
+        $this->getReceiver()->keepalive($envelope, $seconds);
+    }
+
     public function getMessageCount(): int
     {
         return $this->getReceiver()->getMessageCount();

src/Symfony/Component/Messenger/Bridge/Doctrine/composer.json

@@ -18,7 +18,7 @@
     "require": {
         "php": ">=8.2",
         "doctrine/dbal": "^3.6|^4",
-        "symfony/messenger": "^6.4|^7.0",
+        "symfony/messenger": "^7.2",
         "symfony/service-contracts": "^2.5|^3"
     },
     "require-dev": {