[Messenger] Add options to specify SQS queue attributes and tags

Author: TrePe0

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

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.3
+---
+
+ * Add new `queue_attributes` and `queue_tags` options for SQS queue creation
+
 7.2
 ---
 

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/ConnectionTest.php

@@ -13,7 +13,9 @@
 
 use AsyncAws\Core\Exception\Http\HttpException;
 use AsyncAws\Core\Test\ResultMockFactory;
+use AsyncAws\Sqs\Enum\QueueAttributeName;
 use AsyncAws\Sqs\Result\GetQueueUrlResult;
+use AsyncAws\Sqs\Result\QueueExistsWaiter;
 use AsyncAws\Sqs\Result\ReceiveMessageResult;
 use AsyncAws\Sqs\SqsClient;
 use AsyncAws\Sqs\ValueObject\Message;
@@ -385,6 +387,43 @@ public function testKeepaliveWithTooSmallTtl()
         $connection->keepalive('123', 2);
     }
 
+    public function testQueueAttributesAndTags()
+    {
+        $queueName = 'queueName.fifo';
+        $queueAttributes = [
+            QueueAttributeName::MESSAGE_RETENTION_PERIOD => '900',
+            QueueAttributeName::MAXIMUM_MESSAGE_SIZE => '1024',
+        ];
+        $queueTags = ['tag1' => 'value1', 'tag2' => 'value2'];
+
+        $queueExistsWaiter = ResultMockFactory::waiter(QueueExistsWaiter::class, QueueExistsWaiter::STATE_FAILURE);
+        $client = $this->createMock(SqsClient::class);
+        $client->method('queueExists')->willReturn($queueExistsWaiter);
+        $client->expects($this->once())->method('createQueue')->with(['QueueName' => $queueName, 'Attributes' => array_merge($queueAttributes, [QueueAttributeName::FIFO_QUEUE => 'true']), 'tags' => $queueTags]);
+
+        $connection = new Connection(['queue_name' => $queueName, 'queue_attributes' => $queueAttributes, 'queue_tags' => $queueTags], $client);
+
+        $this->expectException(TransportException::class);
+        $connection->setup();
+    }
+
+    public function testQueueAttributesAndTagsFromDsn()
+    {
+        $httpClient = $this->createMock(HttpClientInterface::class);
+
+        $queueName = 'queueName';
+        $queueAttributes = [
+            QueueAttributeName::MESSAGE_RETENTION_PERIOD => '900',
+            QueueAttributeName::MAXIMUM_MESSAGE_SIZE => '1024',
+        ];
+        $queueTags = ['tag1' => 'value1', 'tag2' => 'value2'];
+
+        $this->assertEquals(
+            new Connection(['queue_name' => $queueName, 'queue_attributes' => $queueAttributes, 'queue_tags' => $queueTags], new SqsClient(['region' => 'eu-west-1', 'accessKeyId' => null, 'accessKeySecret' => null], null, $httpClient)),
+            Connection::fromDsn('sqs://default/'.$queueName, ['queue_attributes' => $queueAttributes, 'queue_tags' => $queueTags], $httpClient)
+        );
+    }
+
     private function getMockedQueueUrlResponse(): MockResponse
     {
         if ($this->isAsyncAwsSqsVersion2Installed()) {

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

@@ -46,6 +46,8 @@ class Connection
         'endpoint' => 'https://sqs.eu-west-1.amazonaws.com',
         'region' => 'eu-west-1',
         'queue_name' => 'messages',
+        'queue_attributes' => null,
+        'queue_tags' => null,
         'account' => null,
         'sslmode' => null,
         'debug' => null,
@@ -89,6 +91,8 @@ public function __destruct()
      * * endpoint: absolute URL to the SQS service (Default: https://sqs.eu-west-1.amazonaws.com)
      * * region: name of the AWS region (Default: eu-west-1)
      * * queue_name: name of the queue (Default: messages)
+     * * queue_attributes: attributes of the queue, array<QueueAttributeName::*, string>
+     * * queue_tags: tags of the queue, array<string, string>
      * * account: identifier of the AWS account
      * * access_key: AWS access key
      * * secret_key: AWS secret key
@@ -132,6 +136,8 @@ public static function fromDsn(#[\SensitiveParameter] string $dsn, array $option
             'visibility_timeout' => null !== $options['visibility_timeout'] ? (int) $options['visibility_timeout'] : null,
             'auto_setup' => filter_var($options['auto_setup'], \FILTER_VALIDATE_BOOL),
             'queue_name' => (string) $options['queue_name'],
+            'queue_attributes' => $options['queue_attributes'],
+            'queue_tags' => $options['queue_tags'],
         ];
 
         $clientConfiguration = [
@@ -278,12 +284,14 @@ public function setup(): void
             throw new InvalidArgumentException(\sprintf('The Amazon SQS queue "%s" does not exist (or you don\'t have permissions on it), and can\'t be created when an account is provided.', $this->configuration['queue_name']));
         }
 
-        $parameters = ['QueueName' => $this->configuration['queue_name']];
+        $parameters = [
+            'QueueName' => $this->configuration['queue_name'],
+            'Attributes' => $this->configuration['queue_attributes'],
+            'tags' => $this->configuration['queue_tags'],
+        ];
 
         if (self::isFifoQueue($this->configuration['queue_name'])) {
-            $parameters['Attributes'] = [
-                'FifoQueue' => 'true',
-            ];
+            $parameters['Attributes'][QueueAttributeName::FIFO_QUEUE] = 'true';
         }
 
         $this->client->createQueue($parameters);