Enhancement: Add Keepalive Support to Doctrine Messenger Transport

silasjoisten · silasjoisten · commit e5fcaad58674 · 2025-01-24T09:28:38.000+02:00
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/CHANGELOG.md b/src/Symfony/Component/Messenger/Bridge/Doctrine/CHANGELOG.md
@@ -1,6 +1,10 @@
 CHANGELOG
 =========
 
+7.3
+---
+ * Add "keepalive" support
+
 7.1
 ---
 
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/Connection.php b/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();
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineReceiver.php b/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 <vincent.touzet@gmail.com>
  */
-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) {
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineTransport.php b/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 <vincent.touzet@gmail.com>
  */
-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();
