[Amqp] Add amqps support

Author: Valentin

CHANGELOG.md

@@ -5,6 +5,7 @@ CHANGELOG
 -----
 
  * Add option to confirm message delivery
+ * DSN now support AMQPS out-of-the-box.
 
 5.1.0
 -----

Tests/Transport/ConnectionTest.php

@@ -53,6 +53,24 @@ public function testItCanBeConstructedWithDefaults()
         );
     }
 
+    public function testItCanBeConstructedWithAnAmqpsDsn()
+    {
+        $this->assertEquals(
+            new Connection([
+                'host' => 'localhost',
+                'port' => 5671,
+                'vhost' => '/',
+                'cacert' => '/etc/ssl/certs',
+            ], [
+                'name' => self::DEFAULT_EXCHANGE_NAME,
+            ], [
+                self::DEFAULT_EXCHANGE_NAME => [],
+            ]),
+            Connection::fromDsn('amqps://localhost?'.
+                'cacert=/etc/ssl/certs')
+        );
+    }
+
     public function testItGetsParametersFromTheDsn()
     {
         $this->assertEquals(
@@ -314,6 +332,45 @@ public function testItSetupsTheConnection()
         $connection->publish('body');
     }
 
+    public function testItSetupsTheTTLConnection()
+    {
+        $amqpConnection = $this->createMock(\AMQPConnection::class);
+        $amqpChannel = $this->createMock(\AMQPChannel::class);
+        $amqpExchange = $this->createMock(\AMQPExchange::class);
+        $amqpQueue0 = $this->createMock(\AMQPQueue::class);
+        $amqpQueue1 = $this->createMock(\AMQPQueue::class);
+
+        $factory = $this->createMock(AmqpFactory::class);
+        $factory->method('createConnection')->willReturn($amqpConnection);
+        $factory->method('createChannel')->willReturn($amqpChannel);
+        $factory->method('createExchange')->willReturn($amqpExchange);
+        $factory->method('createQueue')->will($this->onConsecutiveCalls($amqpQueue0, $amqpQueue1));
+
+        $amqpExchange->expects($this->once())->method('declareExchange');
+        $amqpExchange->expects($this->once())->method('publish')->with('body', 'routing_key', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
+        $amqpQueue0->expects($this->once())->method('declareQueue');
+        $amqpQueue0->expects($this->exactly(2))->method('bind')->withConsecutive(
+            [self::DEFAULT_EXCHANGE_NAME, 'binding_key0'],
+            [self::DEFAULT_EXCHANGE_NAME, 'binding_key1']
+        );
+        $amqpQueue1->expects($this->once())->method('declareQueue');
+        $amqpQueue1->expects($this->exactly(2))->method('bind')->withConsecutive(
+            [self::DEFAULT_EXCHANGE_NAME, 'binding_key2'],
+            [self::DEFAULT_EXCHANGE_NAME, 'binding_key3']
+        );
+
+        $dsn = 'amqps://localhost?'.
+            'cacert=/etc/ssl/certs&'.
+            'exchange[default_publish_routing_key]=routing_key&'.
+            'queues[queue0][binding_keys][0]=binding_key0&'.
+            'queues[queue0][binding_keys][1]=binding_key1&'.
+            'queues[queue1][binding_keys][0]=binding_key2&'.
+            'queues[queue1][binding_keys][1]=binding_key3';
+
+        $connection = Connection::fromDsn($dsn, [], $factory);
+        $connection->publish('body');
+    }
+
     public function testBindingArguments()
     {
         $amqpConnection = $this->createMock(\AMQPConnection::class);
@@ -506,6 +563,27 @@ public function testObfuscatePasswordInDsn()
         $connection->channel();
     }
 
+    public function testNoCaCertOnSslConnectionFromDsn()
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('No CA certificate has been provided. Set "amqp.cacert" in your php.ini or pass the "cacert" parameter in the DSN to use SSL. Alternatively, you can use amqp:// to use without SSL.');
+
+        $factory = new TestAmqpFactory(
+            $amqpConnection = $this->createMock(\AMQPConnection::class),
+            $amqpChannel = $this->createMock(\AMQPChannel::class),
+            $amqpQueue = $this->createMock(\AMQPQueue::class),
+            $amqpExchange = $this->createMock(\AMQPExchange::class)
+        );
+
+        $oldCaCertValue = ini_set('amqp.cacert', '');
+
+        try {
+            Connection::fromDsn('amqps://', [], $factory);
+        } finally {
+            ini_set('amqp.cacert', $oldCaCertValue);
+        }
+    }
+
     public function testAmqpStampHeadersAreUsed()
     {
         $factory = new TestAmqpFactory(

Transport/AmqpTransportFactory.php

@@ -29,7 +29,7 @@ public function createTransport(string $dsn, array $options, SerializerInterface
 
     public function supports(string $dsn, array $options): bool
     {
-        return 0 === strpos($dsn, 'amqp://');
+        return 0 === strpos($dsn, 'amqp://') || 0 === strpos($dsn, 'amqps://');
     }
 }
 class_alias(AmqpTransportFactory::class, \Symfony\Component\Messenger\Transport\AmqpExt\AmqpTransportFactory::class);

Transport/Connection.php

@@ -167,20 +167,22 @@ public static function fromDsn(string $dsn, array $options = [], AmqpFactory $am
     {
         if (false === $parsedUrl = parse_url($dsn)) {
             // this is a valid URI that parse_url cannot handle when you want to pass all parameters as options
-            if ('amqp://' !== $dsn) {
+            if (!\in_array($dsn, ['amqp://', 'amqps://'])) {
                 throw new InvalidArgumentException(sprintf('The given AMQP DSN "%s" is invalid.', $dsn));
             }
 
             $parsedUrl = [];
         }
 
+        $useAmqps = 0 === strpos($dsn, 'amqps://');
         $pathParts = isset($parsedUrl['path']) ? explode('/', trim($parsedUrl['path'], '/')) : [];
         $exchangeName = $pathParts[1] ?? 'messages';
         parse_str($parsedUrl['query'] ?? '', $parsedQuery);
+        $port = $useAmqps ? 5671 : 5672;
 
         $amqpOptions = array_replace_recursive([
             'host' => $parsedUrl['host'] ?? 'localhost',
-            'port' => $parsedUrl['port'] ?? 5672,
+            'port' => $parsedUrl['port'] ?? $port,
             'vhost' => isset($pathParts[0]) ? urldecode($pathParts[0]) : '/',
             'exchange' => [
                 'name' => $exchangeName,
@@ -216,6 +218,10 @@ public static function fromDsn(string $dsn, array $options = [], AmqpFactory $am
             return $queueOptions;
         }, $queuesOptions);
 
+        if ($useAmqps && !self::hasCaCertConfigured($amqpOptions)) {
+            throw new InvalidArgumentException('No CA certificate has been provided. Set "amqp.cacert" in your php.ini or pass the "cacert" parameter in the DSN to use SSL. Alternatively, you can use amqp:// to use without SSL.');
+        }
+
         return new self($amqpOptions, $exchangeOptions, $queuesOptions, $amqpFactory);
     }
 
@@ -260,6 +266,11 @@ private static function normalizeQueueArguments(array $arguments): array
         return $arguments;
     }
 
+    private static function hasCaCertConfigured(array $amqpOptions): bool
+    {
+        return (isset($amqpOptions['cacert']) && '' !== $amqpOptions['cacert']) || '' !== ini_get('amqp.cacert');
+    }
+
     /**
      * @throws \AMQPException
      */