feature #37759 [Messenger] - Add option to confirm message delivery in Amqp connection (scyzoryck)

This PR was merged into the 5.2-dev branch.

Discussion
----------

[Messenger] - Add option to confirm message delivery in Amqp connection

| Q             | A
| ------------- | ---
| Branch?       | master
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Tickets       | -
| License       | MIT
| Doc PR        | -

Hello! My first PR here.

Sometimes we need to be sure that messages are delivered to Amqp. To make it work we need to wait for the confirmation from the Amqp server. So let's wait for it confirmation if confirmation timeout is defined. Default behaviour are not modified - waiting for confirmation will impact performance of sending message.

Commits
-------

5682d76b2a Messenger - Add option to confirm message delivery in Amqp connection

Author: fabpot

CHANGELOG.md

@@ -6,3 +6,8 @@ CHANGELOG
 
  * Introduced the AMQP bridge.
  * Deprecated use of invalid options
+
+5.2.0
+-----
+
+ * Add option to confirm message delivery

Tests/Fixtures/long_receiver.php

@@ -13,12 +13,12 @@
 require_once $autoload;
 
 use Symfony\Component\EventDispatcher\EventDispatcher;
+use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpReceiver;
+use Symfony\Component\Messenger\Bridge\Amqp\Transport\Connection;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnSigtermSignalListener;
 use Symfony\Component\Messenger\MessageBusInterface;
-use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpReceiver;
-use Symfony\Component\Messenger\Bridge\Amqp\Transport\Connection;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
 use Symfony\Component\Messenger\Worker;
 use Symfony\Component\Serializer as SerializerComponent;

Tests/Transport/ConnectionTest.php

@@ -625,6 +625,51 @@ public function testItCanPublishWithCustomFlagsAndAttributes()
         $connection = Connection::fromDsn('amqp://localhost', [], $factory);
         $connection->publish('body', ['type' => DummyMessage::class], 0, new AmqpStamp('routing_key', \AMQP_IMMEDIATE, ['delivery_mode' => 2]));
     }
+
+    public function testItPublishMessagesWithoutWaitingForConfirmation()
+    {
+        $factory = new TestAmqpFactory(
+            $amqpConnection = $this->createMock(\AMQPConnection::class),
+            $amqpChannel = $this->createMock(\AMQPChannel::class),
+            $amqpQueue = $this->createMock(\AMQPQueue::class),
+            $amqpExchange = $this->createMock(\AMQPExchange::class)
+        );
+
+        $amqpChannel->expects($this->never())->method('waitForConfirm')->with(0.5);
+
+        $connection = Connection::fromDsn('amqp://localhost', [], $factory);
+        $connection->publish('body');
+    }
+
+    public function testSetChannelToConfirmMessage()
+    {
+        $factory = new TestAmqpFactory(
+            $amqpConnection = $this->createMock(\AMQPConnection::class),
+            $amqpChannel = $this->createMock(\AMQPChannel::class),
+            $amqpQueue = $this->createMock(\AMQPQueue::class),
+            $amqpExchange = $this->createMock(\AMQPExchange::class)
+        );
+
+        $amqpChannel->expects($this->once())->method('confirmSelect');
+        $amqpChannel->expects($this->once())->method('setConfirmCallback');
+        $connection = Connection::fromDsn('amqp://localhost?confirm_timeout=0.5', [], $factory);
+        $connection->setup();
+    }
+
+    public function testItCanPublishAndWaitForConfirmation()
+    {
+        $factory = new TestAmqpFactory(
+            $amqpConnection = $this->createMock(\AMQPConnection::class),
+            $amqpChannel = $this->createMock(\AMQPChannel::class),
+            $amqpQueue = $this->createMock(\AMQPQueue::class),
+            $amqpExchange = $this->createMock(\AMQPExchange::class)
+        );
+
+        $amqpChannel->expects($this->once())->method('waitForConfirm')->with(0.5);
+
+        $connection = Connection::fromDsn('amqp://localhost?confirm_timeout=0.5', [], $factory);
+        $connection->publish('body');
+    }
 }
 
 class TestAmqpFactory extends AmqpFactory

Transport/Connection.php

@@ -50,6 +50,7 @@ class Connection
         'heartbeat',
         'read_timeout',
         'write_timeout',
+        'confirm_timeout',
         'connect_timeout',
         'cacert',
         'cert',
@@ -128,6 +129,7 @@ public function __construct(array $connectionOptions, array $exchangeOptions, ar
      *   * read_timeout: Timeout in for income activity. Note: 0 or greater seconds. May be fractional.
      *   * write_timeout: Timeout in for outcome activity. Note: 0 or greater seconds. May be fractional.
      *   * connect_timeout: Connection timeout. Note: 0 or greater seconds. May be fractional.
+     *   * confirm_timeout: Timeout in seconds for confirmation, if none specified transport will not wait for message confirmation. Note: 0 or greater seconds. May be fractional.
      *   * queues[name]: An array of queues, keyed by the name
      *     * binding_keys: The binding keys (if any) to bind to this queue
      *     * binding_arguments: Arguments to be used while binding the queue.
@@ -325,6 +327,10 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
             $amqpStamp ? $amqpStamp->getFlags() : \AMQP_NOPARAM,
             $attributes
         );
+
+        if ('' !== ($this->connectionOptions['confirm_timeout'] ?? '')) {
+            $this->channel()->waitForConfirm((float) $this->connectionOptions['confirm_timeout']);
+        }
     }
 
     private function setupDelay(int $delay, ?string $routingKey)
@@ -478,6 +484,18 @@ public function channel(): \AMQPChannel
             if (isset($this->connectionOptions['prefetch_count'])) {
                 $this->amqpChannel->setPrefetchCount($this->connectionOptions['prefetch_count']);
             }
+
+            if ('' !== ($this->connectionOptions['confirm_timeout'] ?? '')) {
+                $this->amqpChannel->confirmSelect();
+                $this->amqpChannel->setConfirmCallback(
+                    static function (): bool {
+                        return false;
+                    },
+                    static function (): bool {
+                        return false;
+                    }
+                );
+            }
         }
 
         return $this->amqpChannel;