Messenger - Add option to confirm message delivery in Amqp connection

Author: scyzoryck

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;