Merge pull request #98 from Cafeine42/master

Improve SQS Transport Resolver to match standard configuration

Author: mnapoli

src/Resources/config/services.yaml

@@ -22,7 +22,7 @@ services:
     # SQS
     Bref\Symfony\Messenger\Service\Sqs\SqsTransportNameResolver:
         arguments:
-            - '@Bref\Symfony\Messenger\Service\MessengerTransportConfiguration'
+            $messengerTransportsConfiguration: '%messenger.transports%'
 
     # EventBridge
     Bref\Symfony\Messenger\Service\EventBridge\EventBridgeTransportFactory:

src/Service/Sqs/SqsTransportNameResolver.php

@@ -1,9 +1,8 @@
-<?php
+<?php declare(strict_types=1);
 
 namespace Bref\Symfony\Messenger\Service\Sqs;
 
 use Bref\Event\Sqs\SqsRecord;
-use Bref\Symfony\Messenger\Service\MessengerTransportConfiguration;
 use InvalidArgumentException;
 
 /** @final */
@@ -12,18 +11,45 @@ class SqsTransportNameResolver
     private const TRANSPORT_PROTOCOL = 'sqs://';
 
     public function __construct(
-        private MessengerTransportConfiguration $configurationProvider
+        private array $messengerTransportsConfiguration
     ) {
     }
 
     public function __invoke(SqsRecord $sqsRecord): string
     {
-        if (!array_key_exists('eventSourceARN', $sqsRecord->toArray())) {
+        if (! array_key_exists('eventSourceARN', $sqsRecord->toArray())) {
             throw new InvalidArgumentException('EventSourceArn is missing in sqs record.');
         }
 
         $eventSourceArn = $sqsRecord->toArray()['eventSourceARN'];
+        $eventSourceWithProtocol = self::TRANSPORT_PROTOCOL . $eventSourceArn;
 
-        return $this->configurationProvider->provideTransportFromEventSource(self::TRANSPORT_PROTOCOL . $eventSourceArn);
+        foreach ($this->messengerTransportsConfiguration as $messengerTransport => $messengerOptions) {
+            $dsn = $this->extractDsnFromTransport($messengerOptions);
+
+            if ($dsn === $eventSourceWithProtocol) {
+                return $messengerTransport;
+            }
+
+            // Rebuild SQS ARN from https://sqs.eu-west-3.amazonaws.com/0123456789/messages?key=value
+            if (preg_match('/^https:\/\/sqs\.([^.]+)\.amazonaws\.com\/([^\/]+)\/([^?]+)/', (string) $dsn, $matches)) {
+                $arn = 'arn:aws:sqs:' . $matches[1] . ':' . $matches[2] . ':' . $matches[3];
+
+                if ($eventSourceWithProtocol === 'sqs://' . $arn) {
+                    return $messengerTransport;
+                }
+            }
+        }
+
+        throw new InvalidArgumentException(sprintf('No transport found for eventSource "%s".', $eventSourceWithProtocol));
+    }
+
+    private function extractDsnFromTransport(string|array $messengerTransport): string
+    {
+        if (is_array($messengerTransport) && array_key_exists('dsn', $messengerTransport)) {
+            return $messengerTransport['dsn'];
+        }
+
+        return $messengerTransport;
     }
 }

tests/Unit/Service/Sqs/SqsTransportNameResolverTest.php

@@ -0,0 +1,89 @@
+<?php declare(strict_types=1);
+
+namespace Bref\Symfony\Messenger\Test\Unit\Service\Sqs;
+
+use Bref\Event\Sqs\SqsRecord;
+use Bref\Symfony\Messenger\Service\Sqs\SqsTransportNameResolver;
+use InvalidArgumentException;
+use PHPUnit\Framework\TestCase;
+use Prophecy\PhpUnit\ProphecyTrait;
+
+final class SqsTransportNameResolverTest extends TestCase
+{
+    use ProphecyTrait;
+
+    public function test_event_source_can_resolved_as_expected(): void
+    {
+        $messengerTransportsConfiguration = [
+            'async' => [
+                'dsn' => 'sqs://arn:aws:sqs:us-east-1:1234567890:some-queue-name',
+            ],
+        ];
+
+        $transportNameResolver = new SqsTransportNameResolver($messengerTransportsConfiguration);
+
+        $event = new SqsRecord([
+            'messageId' => '19dd0b57-b21e-4ac1-bd88-01bbb068cb78',
+            'body' => 'Test message.',
+            'messageAttributes' => [],
+            'attributes' => [
+                'ApproximateReceiveCount' => '1',
+            ],
+            'receiptHandle' => 'AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...',
+            'eventSource' => 'aws:sqs',
+            'eventSourceARN' => 'arn:aws:sqs:us-east-1:1234567890:some-queue-name',
+        ]);
+
+        self::assertSame('async', ($transportNameResolver)($event));
+    }
+
+    public function test_event_source_can_resolved_as_expected_with_queue_url(): void
+    {
+        $messengerTransportsConfiguration = [
+            'async' => [
+                'dsn' => 'https://sqs.us-east-1.amazonaws.com/1234567890/some-queue-name',
+            ],
+        ];
+
+        $transportNameResolver = new SqsTransportNameResolver($messengerTransportsConfiguration);
+
+        $event = new SqsRecord([
+            'messageId' => '19dd0b57-b21e-4ac1-bd88-01bbb068cb78',
+            'body' => 'Test message.',
+            'messageAttributes' => [],
+            'attributes' => [
+                'ApproximateReceiveCount' => '1',
+            ],
+            'receiptHandle' => 'AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...',
+            'eventSource' => 'aws:sqs',
+            'eventSourceARN' => 'arn:aws:sqs:us-east-1:1234567890:some-queue-name',
+        ]);
+
+        self::assertEquals('async', ($transportNameResolver)($event));
+    }
+
+    public function test_throws_exception_if_event_source_arn_does_not_exist(): void
+    {
+        $messengerTransportsConfiguration = [
+            'transport1' => [
+                'dsn' => 'sqs://arn:aws:sqs:us-east-1:0123456789:some-queue-name',
+            ],
+        ];
+
+        $transportNameResolver = new SqsTransportNameResolver($messengerTransportsConfiguration);
+
+        $eventWithMissingeventSourceARN = new SqsRecord([
+            'messageId' => '19dd0b57-b21e-4ac1-bd88-01bbb068cb78',
+            'body' => 'Test message.',
+            'messageAttributes' => [],
+            'attributes' => [
+                'ApproximateReceiveCount' => '1',
+            ],
+            'receiptHandle' => 'AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...',
+            'eventSource' => 'aws:sqs',
+        ]);
+
+        $this->expectException(InvalidArgumentException::class);
+        ($transportNameResolver)($eventWithMissingeventSourceARN);
+    }
+}