Messenger - Add option to confirm message delivery in Amqp connection

scyzoryck · scyzoryck · commit b542b7f12e5b · 2020-08-10T23:52:18.000+02:00
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/CHANGELOG.md b/src/Symfony/Component/Messenger/Bridge/Amqp/CHANGELOG.md
@@ -6,3 +6,4 @@ CHANGELOG
 
  * Introduced the AMQP bridge.
  * Deprecated use of invalid options
+ * Add option to confirm message delivery
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php
@@ -623,6 +623,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
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php b/src/Symfony/Component/Messenger/Bridge/Amqp/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.
@@ -324,6 +326,10 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
             $amqpStamp ? $amqpStamp->getFlags() : AMQP_NOPARAM,
             $attributes
         );
+
+        if (isset($this->connectionOptions['confirm_timeout'])) {
+            $this->channel()->waitForConfirm($this->connectionOptions['confirm_timeout']);
+        }
     }
 
     private function setupDelay(int $delay, ?string $routingKey)
@@ -477,6 +483,18 @@ public function channel(): \AMQPChannel
             if (isset($this->connectionOptions['prefetch_count'])) {
                 $this->amqpChannel->setPrefetchCount($this->connectionOptions['prefetch_count']);
             }
+
+            if (isset($this->connectionOptions['confirm_timeout'])) {
+                $this->amqpChannel->confirmSelect();
+                $this->amqpChannel->setConfirmCallback(
+                    static function (): bool {
+                        return false;
+                    },
+                    static function (): bool {
+                        return false;
+                    }
+                );
+            }
         }
 
         return $this->amqpChannel;

Original file line number	Diff line number	Diff line change
`@@ -6,3 +6,4 @@ CHANGELOG`
`6`	`6`
`7`	`7`	`* Introduced the AMQP bridge.`
`8`	`8`	`* Deprecated use of invalid options`
	`9`	`+ * Add option to confirm message delivery`