Removed BC break and moved $transportName to correct place again

Author: robinlehrmann

src/Service/Sns/SnsConsumer.php

@@ -5,7 +5,9 @@
 use Bref\Context\Context;
 use Bref\Event\Sns\SnsEvent;
 use Bref\Event\Sns\SnsHandler;
+use Bref\Event\Sns\SnsRecord;
 use Bref\Symfony\Messenger\Service\BusDriver;
+use LogicException;
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 
@@ -15,19 +17,19 @@ final class SnsConsumer extends SnsHandler
     private $bus;
     /** @var SerializerInterface */
     protected $serializer;
-    /** @var SnsTransportNameResolver */
-    private $transportNameResolver;
-    /** @var BusDriver */
-    private $busDriver;
     /** @var string|null */
     private $transportName;
+    /** @var BusDriver */
+    private $busDriver;
+    /** @var SnsTransportNameResolver|null */
+    private $transportNameResolver;
 
     public function __construct(
         BusDriver $busDriver,
         MessageBusInterface $bus,
         SerializerInterface $serializer,
-        SnsTransportNameResolver $transportNameResolver,
-        string $transportName = null
+        string $transportName = null,
+        SnsTransportNameResolver $transportNameResolver = null,
     ) {
         $this->busDriver = $busDriver;
         $this->bus = $bus;
@@ -43,7 +45,16 @@ public function handleSns(SnsEvent $event, Context $context): void
             $headers = isset($attributes['Headers']) ? $attributes['Headers']->getValue() : '[]';
             $envelope = $this->serializer->decode(['body' => $record->getMessage(), 'headers' => json_decode($headers, true)]);
 
-            $this->busDriver->putEnvelopeOnBus($this->bus, $envelope, $this->transportName ?? ($this->transportNameResolver)($record));
+            $this->busDriver->putEnvelopeOnBus($this->bus, $envelope, $this->resolveTransportName($record));
         }
     }
+
+    private function resolveTransportName(SnsRecord $record): string
+    {
+        if (null === $this->transportName && null === $this->transportNameResolver) {
+            throw new LogicException('You need to set $transportNameResolver or $transportName.');
+        }
+
+        return $this->transportName ?? ($this->transportNameResolver)($record);
+    }
 }

src/Service/Sqs/SqsConsumer.php

@@ -7,6 +7,7 @@
 use Bref\Event\Sqs\SqsHandler;
 use Bref\Event\Sqs\SqsRecord;
 use Bref\Symfony\Messenger\Service\BusDriver;
+use LogicException;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsReceivedStamp;
@@ -38,18 +39,18 @@ public function __construct(
         BusDriver $busDriver,
         MessageBusInterface $bus,
         SerializerInterface $serializer,
-        SqsTransportNameResolver $transportNameResolver,
         string $transportName = null,
         LoggerInterface $logger = null,
-        bool $partialBatchFailure = false
+        bool $partialBatchFailure = false,
+        SqsTransportNameResolver $transportNameResolver = null
     ) {
         $this->busDriver = $busDriver;
         $this->bus = $bus;
         $this->serializer = $serializer;
-        $this->transportNameResolver = $transportNameResolver;
         $this->transportName = $transportName;
         $this->logger = $logger ?? new NullLogger();
         $this->partialBatchFailure = $partialBatchFailure;
+        $this->transportNameResolver = $transportNameResolver;
     }
 
     public function handleSqs(SqsEvent $event, Context $context): void
@@ -97,7 +98,7 @@ public function handleSqs(SqsEvent $event, Context $context): void
                 if ('' !== $context->getTraceId()) {
                     $stamps[] = new AmazonSqsXrayTraceHeaderStamp($context->getTraceId());
                 }
-                $this->busDriver->putEnvelopeOnBus($this->bus, $envelope->with(...$stamps), $this->transportName ?? ($this->transportNameResolver)($record));
+                $this->busDriver->putEnvelopeOnBus($this->bus, $envelope->with(...$stamps), $this->resolveTransportName($record));
             } catch (UnrecoverableExceptionInterface $exception) {
                 $this->logger->error(sprintf('SQS record with id "%s" failed to be processed. But failure was marked as unrecoverable. Message will be acknowledged.', $record->getMessageId()));
                 $this->logger->error($exception);
@@ -120,4 +121,13 @@ private function readMessageGroupIdOfRecord(SqsRecord $record): ?string
         $recordAsArray = $record->toArray();
         return $recordAsArray['attributes']['MessageGroupId'] ?? null;
     }
+
+    private function resolveTransportName(SqsRecord $record): string
+    {
+        if (null === $this->transportName && null === $this->transportNameResolver) {
+            throw new LogicException('You need to set $transportNameResolver or $transportName.');
+        }
+
+        return $this->transportName ?? ($this->transportNameResolver)($record);
+    }
 }

tests/Unit/Service/Sns/SnsConsumerTest.php

@@ -7,6 +7,7 @@
 use Bref\Symfony\Messenger\Service\BusDriver;
 use Bref\Symfony\Messenger\Service\Sns\SnsConsumer;
 use Bref\Symfony\Messenger\Service\Sns\SnsTransportNameResolver;
+use LogicException;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\MessageBus;
@@ -60,6 +61,7 @@ public function test_serializer()
             $this->busDriver,
             $this->bus,
             $this->serializer,
+            null,
             $this->snsTransportNameResolver
         );
 
@@ -86,6 +88,7 @@ public function test_event_with_transport_detection(): void
             $this->busDriver,
             $this->bus,
             $this->serializer,
+            null,
             $this->snsTransportNameResolver
         );
 
@@ -111,8 +114,8 @@ public function test_event_with_manually_set_transport(): void
             $this->busDriver,
             $this->bus,
             $this->serializer,
-            $this->snsTransportNameResolver,
-            'async'
+            'async',
+            $this->snsTransportNameResolver
         );
 
         $event = $this->snsEvent($body, $headers);

tests/Unit/Service/Sqs/SqsConsumerTest.php

@@ -63,7 +63,15 @@ public function test_batch_events()
             ),
         ];
 
-        $consumer = new SqsConsumer($this->busDriver->reveal(), $this->bus, $this->serializer->reveal(), $this->sqsTransportNameResolver->reveal());
+        $consumer = new SqsConsumer(
+            $this->busDriver->reveal(),
+            $this->bus,
+            $this->serializer->reveal(),
+            null,
+            null,
+            false,
+            $this->sqsTransportNameResolver->reveal()
+        );
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));
         $this->assertEmpty($failures);
     }
@@ -84,6 +92,9 @@ public function test_event_with_transport_detection(): void
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
+            null,
+            null,
+            false,
             $this->sqsTransportNameResolver->reveal()
         );
 
@@ -108,8 +119,10 @@ public function test_event_with_manually_set_transport(): void
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
+            'async_test',
+            null,
+            false,
             $this->sqsTransportNameResolver->reveal(),
-            'async_test'
         );
 
         $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));
@@ -143,6 +156,9 @@ public function test_batch_events_with_failure()
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
+            $transport,
+            null,
+            false,
             $this->sqsTransportNameResolver->reveal()
         );
 
@@ -178,10 +194,10 @@ public function test_batch_events_failure_with_partial_batch_failure_enabled()
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
-            $this->sqsTransportNameResolver->reveal(),
             null,
             null,
-            true
+            true,
+            $this->sqsTransportNameResolver->reveal()
         );
 
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));
@@ -221,10 +237,10 @@ public function test_batch_events_failure_on_fifo_queue_with_partial_batch_failu
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
-            $this->sqsTransportNameResolver->reveal(),
-            null,
+            $transport,
             null,
-            true
+            true,
+            $this->sqsTransportNameResolver->reveal()
         );
 
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));
@@ -266,6 +282,9 @@ public function test_x_ray_header_is_dispatched_on_bus()
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
+            $transport,
+            null,
+            false,
             $this->sqsTransportNameResolver->reveal()
         );
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', $xrayTraceId));
@@ -305,10 +324,10 @@ public function test_unrecoverable_exception_during_batch()
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
-            $this->sqsTransportNameResolver->reveal(),
-            null,
+            $transport,
             null,
-            true
+            true,
+            $this->sqsTransportNameResolver->reveal(),
         );
 
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));
@@ -356,10 +375,10 @@ public function test_message_group_id_during_batch_of_fifo_queue()
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
-            $this->sqsTransportNameResolver->reveal(),
-            null,
+            $transport,
             null,
-            true
+            true,
+            $this->sqsTransportNameResolver->reveal(),
         );
 
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));
@@ -410,10 +429,10 @@ public function test_different_message_group_id_failed_during_batch_of_fifo_queu
             $this->busDriver->reveal(),
             $this->bus,
             $this->serializer->reveal(),
-            $this->sqsTransportNameResolver->reveal(),
-            null,
+            $transport,
             null,
-            true
+            true,
+            $this->sqsTransportNameResolver->reveal(),
         );
 
         $failures = $consumer->handle(['Records' => $sqsRecords], new Context('', 0, '', ''));