Remove DoctrineClearEntityManagerWorkerSubscriber during process

src/Transport/TestTransport.php

@@ -3,6 +3,7 @@
 namespace Zenstruck\Messenger\Test\Transport;
 
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
 use Symfony\Component\Messenger\Event\WorkerRunningEvent;
@@ -32,6 +33,9 @@ final class TestTransport implements TransportInterface
     private MessageBusInterface $bus;
     private SerializerInterface $serializer;
 
+    /** @var EventSubscriberInterface[] */
+    private array $subscribersToRemove;
+
     /** @var array<string, bool> */
     private static array $intercept = [];
 
@@ -59,16 +63,24 @@ final class TestTransport implements TransportInterface
     /**
      * @internal
      *
-     * @param array<string,bool> $options
+     * @param EventSubscriberInterface[] $subscribersToRemove
+     * @param array<string,bool>         $options
      */
-    public function __construct(string $name, MessageBusInterface $bus, EventDispatcherInterface $dispatcher, SerializerInterface $serializer, array $options = [])
-    {
+    public function __construct(
+        string $name,
+        MessageBusInterface $bus,
+        EventDispatcherInterface $dispatcher,
+        SerializerInterface $serializer,
+        array $subscribersToRemove,
+        array $options = []
+    ) {
         $options = \array_merge(self::DEFAULT_OPTIONS, $options);
 
         $this->name = $name;
         $this->dispatcher = $dispatcher;
         $this->bus = $bus;
         $this->serializer = $serializer;
+        $this->subscribersToRemove = $subscribersToRemove;
 
         self::$intercept[$name] ??= $options['intercept'];
         self::$catchExceptions[$name] ??= $options['catch_exceptions'];
@@ -131,6 +143,11 @@ public function process(int $number = -1): self
         $listeners = [];
         $subscribers = [];
 
+        // remove default subscribers
+        foreach ($this->subscribersToRemove as $subscriber) {
+            $this->dispatcher->removeSubscriber($subscriber);
+        }
+
         $this->dispatcher->addListener(
             WorkerRunningEvent::class,
             $listeners[WorkerRunningEvent::class] = static function(WorkerRunningEvent $event) use (&$processCount) {
@@ -171,6 +188,11 @@ public function process(int $number = -1): self
             $this->dispatcher->removeSubscriber($subscriber);
         }
 
+        // re-add default subscribers
+        foreach ($this->subscribersToRemove as $subscriber) {
+            $this->dispatcher->addSubscriber($subscriber);
+        }
+
         if ($number > 0) {
             Assert::that($processCount)->is($number, 'Expected to process {expected} messages but only processed {actual}.');
         }

src/Transport/TestTransportFactory.php

@@ -3,6 +3,7 @@
 namespace Zenstruck\Messenger\Test\Transport;
 
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
@@ -18,6 +19,9 @@ final class TestTransportFactory implements TransportFactoryInterface
     private MessageBusInterface $bus;
     private EventDispatcherInterface $dispatcher;
 
+    /** @var EventSubscriberInterface[] */
+    private array $subscribersToRemove = [];
+
     public function __construct(MessageBusInterface $bus, EventDispatcherInterface $dispatcher)
     {
         $this->bus = $bus;
@@ -26,14 +30,26 @@ public function __construct(MessageBusInterface $bus, EventDispatcherInterface $
 
     public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface // @phpstan-ignore-line
     {
-        return new TestTransport($options['transport_name'], $this->bus, $this->dispatcher, $serializer, $this->parseDsn($dsn));
+        return new TestTransport(
+            $options['transport_name'],
+            $this->bus,
+            $this->dispatcher,
+            $serializer,
+            $this->subscribersToRemove,
+            $this->parseDsn($dsn)
+        );
     }
 
     public function supports(string $dsn, array $options): bool // @phpstan-ignore-line
     {
         return 0 === \mb_strpos($dsn, 'test://');
     }
 
+    public function registerSubscriberToRemove(EventSubscriberInterface $subscriber): void
+    {
+        $this->subscribersToRemove[] = $subscriber;
+    }
+
     /**
      * @return array<string,bool>
      */

src/ZenstruckMessengerTestBundle.php

@@ -49,5 +49,12 @@ public function process(ContainerBuilder $container): void
 
             $registry->addMethodCall('register', [$name, new Reference($id)]);
         }
+
+        $factory = $container->getDefinition('zenstruck_messenger_test.transport_factory');
+
+        if ($container->hasDefinition($id = 'doctrine.orm.messenger.event_subscriber.doctrine_clear_entity_manager')) {
+            // we want to remove this subscriber so the em isn't reset when processing messages
+            $factory->addMethodCall('registerSubscriberToRemove', [new Reference($id)]);
+        }
     }
 }