feature symfony#59862 [Messenger] Allow to close the transport connection (andrew-demb)

This PR was squashed before being merged into the 7.3 branch.

Discussion
----------

[Messenger] Allow to close the transport connection

| Q             | A
| ------------- | ---
| Branch?       | 7.3
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Issues        | Fix symfony#53543
| License       | MIT

<!--
Replace this notice by a description of your feature/bugfix.
This will help reviewers and should be a good start for the documentation.

Additionally (see https://symfony.com/releases):
 - Always add tests and ensure they pass.
 - Bug fixes must be submitted against the lowest maintained branch where they apply
   (lowest branches are regularly merged to upper ones so they get the fixes too).
 - Features and deprecations must be submitted against the latest branch.
 - For new features, provide some code snippets to help understand usage.
 - Changelog entry should follow https://symfony.com/doc/current/contributing/code/conventions.html#writing-a-changelog-entry
 - Never break backward compatibility (see https://symfony.com/bc).
-->

~~1. Implemented the possibility to make messenger transports resettable~~
~~2. Implemented reset Redis connection for Redis messenger transport~~

~~This feature may lead to decreased performance for messenger consumers (because connection will be resetted between processing messages).~~

~~One way to resolve it that I found - add configuration for resettable transports - aka "reset connection on kernel reset: true/false". For consumers it can be configured via env var to be "false", but for web "true".~~

----

**UPD 2025-02-27**: According to the feedback, I changed the implementation to another one to help with our use case.

Implemented a way to close messenger transport from the application, to allow free resources for long-running processes (like a long-running webserver)

Commits
-------

9788aee [Messenger] Allow to close the transport connection

Author: fabpot

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

@@ -4,6 +4,7 @@ CHANGELOG
 7.3
 ---
 
+ * Implement the `CloseableTransportInterface` to allow closing the transport
  * Add new `queue_attributes` and `queue_tags` options for SQS queue creation
 
 7.2

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsTransport.php

@@ -14,6 +14,7 @@
 use AsyncAws\Core\Exception\Http\HttpException;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\TransportException;
+use Symfony\Component\Messenger\Transport\CloseableTransportInterface;
 use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
@@ -27,7 +28,7 @@
 /**
  * @author Jérémy Derussé <[email protected]>
  */
-class AmazonSqsTransport implements TransportInterface, KeepaliveReceiverInterface, SetupableTransportInterface, MessageCountAwareInterface, ResetInterface
+class AmazonSqsTransport implements TransportInterface, KeepaliveReceiverInterface, SetupableTransportInterface, CloseableTransportInterface, MessageCountAwareInterface, ResetInterface
 {
     private SerializerInterface $serializer;
 
@@ -91,6 +92,11 @@ public function reset(): void
         }
     }
 
+    public function close(): void
+    {
+        $this->reset();
+    }
+
     private function getReceiver(): MessageCountAwareInterface&ReceiverInterface
     {
         return $this->receiver ??= new AmazonSqsReceiver($this->connection, $this->serializer);

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

@@ -19,7 +19,7 @@
         "php": ">=8.2",
         "async-aws/core": "^1.7",
         "async-aws/sqs": "^1.0|^2.0",
-        "symfony/messenger": "^7.2",
+        "symfony/messenger": "^7.3",
         "symfony/service-contracts": "^2.5|^3",
         "psr/log": "^1|^2|^3"
     },

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

@@ -4,6 +4,7 @@ CHANGELOG
 7.1
 ---
 
+* Implement the `CloseableTransportInterface` to allow closing the AMQP connection
  * Add option `delay[arguments]` in the transport definition
 
 6.0

src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpTransport.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger\Bridge\Amqp\Transport;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\CloseableTransportInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
@@ -22,7 +23,7 @@
 /**
  * @author Nicolas Grekas <[email protected]>
  */
-class AmqpTransport implements QueueReceiverInterface, TransportInterface, SetupableTransportInterface, MessageCountAwareInterface
+class AmqpTransport implements QueueReceiverInterface, TransportInterface, SetupableTransportInterface, CloseableTransportInterface, MessageCountAwareInterface
 {
     private SerializerInterface $serializer;
     private AmqpReceiver $receiver;
@@ -70,6 +71,11 @@ public function getMessageCount(): int
         return $this->getReceiver()->getMessageCount();
     }
 
+    public function close(): void
+    {
+        $this->connection->clear();
+    }
+
     private function getReceiver(): AmqpReceiver
     {
         return $this->receiver ??= new AmqpReceiver($this->connection, $this->serializer);

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

@@ -551,7 +551,7 @@ private function clearWhenDisconnected(): void
         }
     }
 
-    private function clear(): void
+    public function clear(): void
     {
         unset($this->amqpChannel, $this->amqpExchange, $this->amqpDelayExchange);
         $this->amqpQueues = [];

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

@@ -18,7 +18,7 @@
     "require": {
         "php": ">=8.2",
         "ext-amqp": "*",
-        "symfony/messenger": "^6.4|^7.0"
+        "symfony/messenger": "^7.3"
     },
     "require-dev": {
         "symfony/event-dispatcher": "^6.4|^7.0",

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

@@ -4,6 +4,7 @@ CHANGELOG
 7.3
 ---
 
+ * Implement the `CloseableTransportInterface` to allow closing the Redis connection
  * Implement the `KeepaliveReceiverInterface` to enable asynchronously notifying Redis that the job is still being processed, in order to avoid timeouts
 
 6.3

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

@@ -55,7 +55,8 @@ class Connection
         'ssl' => null, // see https://php.net/context.ssl
     ];
 
-    private \Redis|Relay|\RedisCluster|\Closure $redis;
+    private \Redis|Relay|\RedisCluster|null $redis = null;
+    private \Closure $redisInitializer;
     private string $stream;
     private string $queue;
     private string $group;
@@ -112,9 +113,9 @@ public function __construct(array $options, \Redis|Relay|\RedisCluster|null $red
 
         if ((\is_array($host) && null === $sentinelMaster) || $redis instanceof \RedisCluster) {
             $hosts = \is_string($host) ? [$host.':'.$port] : $host; // Always ensure we have an array
-            $this->redis = static fn () => self::initializeRedisCluster($redis, $hosts, $auth, $options);
+            $this->redisInitializer = static fn () => self::initializeRedisCluster($redis, $hosts, $auth, $options);
         } else {
-            $this->redis = static function () use ($redis, $sentinelMaster, $host, $port, $options, $auth) {
+            $this->redisInitializer = static function () use ($redis, $sentinelMaster, $host, $port, $options, $auth) {
                 if (null !== $sentinelMaster) {
                     $sentinelClass = \extension_loaded('redis') ? \RedisSentinel::class : Sentinel::class;
                     $hostIndex = 0;
@@ -737,10 +738,15 @@ private function rawCommand(string $command, ...$arguments): mixed
 
     private function getRedis(): \Redis|Relay|\RedisCluster
     {
-        if ($this->redis instanceof \Closure) {
-            $this->redis = ($this->redis)();
+        if (!$this->redis) {
+            $this->redis = ($this->redisInitializer)();
         }
 
         return $this->redis;
     }
+
+    public function close(): void
+    {
+        $this->redis = null;
+    }
 }

src/Symfony/Component/Messenger/Bridge/Redis/Transport/RedisTransport.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger\Bridge\Redis\Transport;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\CloseableTransportInterface;
 use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
@@ -23,7 +24,7 @@
  * @author Alexander Schranz <[email protected]>
  * @author Antoine Bluchet <[email protected]>
  */
-class RedisTransport implements TransportInterface, KeepaliveReceiverInterface, SetupableTransportInterface, MessageCountAwareInterface
+class RedisTransport implements TransportInterface, KeepaliveReceiverInterface, SetupableTransportInterface, CloseableTransportInterface, MessageCountAwareInterface
 {
     private SerializerInterface $serializer;
     private RedisReceiver $receiver;
@@ -71,6 +72,11 @@ public function getMessageCount(): int
         return $this->getReceiver()->getMessageCount();
     }
 
+    public function close(): void
+    {
+        $this->connection->close();
+    }
+
     private function getReceiver(): RedisReceiver
     {
         return $this->receiver ??= new RedisReceiver($this->connection, $this->serializer);