Merge pull request #2 from tienvx/filter-envelope-by-transport

Allow filter envelope by transport

src/EventListener/SendMessageToTransportsEventListener.php

@@ -13,6 +13,9 @@ public function __construct(private EnvelopeCollectorInterface $collector)
 
     public function __invoke(SendMessageToTransportsEvent $event): void
     {
-        $this->collector->collect($event->getEnvelope());
+        $this->collector->collect(
+            $event->getEnvelope(),
+            method_exists($event, 'getSenders') ? array_keys($event->getSenders()) : null
+        );
     }
 }

src/Exception/LogicException.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Tienvx\Bundle\PactMessengerBundle\Exception;
+
+class LogicException extends \LogicException
+{
+}

src/Service/EnvelopeCollector.php

@@ -3,26 +3,41 @@
 namespace Tienvx\Bundle\PactMessengerBundle\Service;
 
 use Symfony\Component\Messenger\Envelope;
+use Tienvx\Bundle\PactMessengerBundle\Exception\LogicException;
 
 class EnvelopeCollector implements EnvelopeCollectorInterface
 {
     /** @var array<int, Envelope> */
     private array $envelopes;
 
-    public function collect(Envelope $envelope): void
+    /** @var array<int, string[]|null> */
+    private array $transportsByEnvelope;
+
+    public function collect(Envelope $envelope, ?array $transports = null): void
     {
         $this->envelopes[] = $envelope;
+        $this->transportsByEnvelope[spl_object_id($envelope)] = $transports;
     }
 
     public function getAll(): array
     {
         return $this->envelopes;
     }
 
-    public function getSingle(string $messageFqcn): ?Envelope
+    public function getSingle(string $messageFqcn, ?string $transport = null): ?Envelope
     {
         foreach ($this->envelopes as $envelope) {
-            if ($envelope->getMessage() instanceof $messageFqcn) {
+            if (!$envelope->getMessage() instanceof $messageFqcn) {
+                continue;
+            }
+            if (is_null($transport)) {
+                return $envelope;
+            }
+            $transports = $this->transportsByEnvelope[spl_object_id($envelope)];
+            if (is_null($transports)) {
+                throw new LogicException('You need to upgrade to Symfony >=6.2 to be able to filter envelope by transport.');
+            }
+            if (in_array($transport, $transports)) {
                 return $envelope;
             }
         }

src/Service/EnvelopeCollectorInterface.php

@@ -6,9 +6,9 @@
 
 interface EnvelopeCollectorInterface
 {
-    public function collect(Envelope $envelope): void;
+    public function collect(Envelope $envelope, ?array $transports = null): void;
 
     public function getAll(): array;
 
-    public function getSingle(string $messageFqcn): ?Envelope;
+    public function getSingle(string $messageFqcn, ?string $transport = null): ?Envelope;
 }