[Messenger] move resetting services at worker stopped into ResetServi…

…cesListener
symfony · fabpot · Jun 12, 2022 · Jun 11, 2022 · Jun 11, 2022 · 16cf4fc4d0617df32cdeb334d2f791f711539a09
commit 16cf4fc4d0617df32cdeb334d2f791f711539a09
diff --git a/src/Symfony/Component/Messenger/EventListener/ResetServicesListener.php b/src/Symfony/Component/Messenger/EventListener/ResetServicesListener.php
@@ -14,6 +14,8 @@
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
 use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
+use Symfony\Contracts\Service\ResetInterface;
 
 /**
  * @author Grégoire Pineau <lyrixx@lyrixx.info>
@@ -34,10 +36,16 @@ public function resetServices(WorkerRunningEvent $event): void
         }
     }
 
+    public function resetServicesAtStop(WorkerStoppedEvent $event): void
+    {
+        $this->servicesResetter->reset();
+    }
+
     public static function getSubscribedEvents(): array
     {
         return [
             WorkerRunningEvent::class => ['resetServices', -1024],
+            WorkerStoppedEvent::class => ['resetServicesAtStop', -1024],
         ];
     }
 }
diff --git a/src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php b/src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php
@@ -26,6 +26,7 @@
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Messenger\RoutableMessageBus;
 use Symfony\Component\Messenger\Stamp\BusNameStamp;
+use Symfony\Component\Messenger\Tests\ResettableDummyReceiver;
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 
 class ConsumeMessagesCommandTest extends TestCase
@@ -116,15 +117,11 @@ public function testRunWithResetServicesOption(bool $shouldReset)
     {
         $envelope = new Envelope(new \stdClass());
 
-        $receiver = $this->createMock(ReceiverInterface::class);
-        $receiver
-            ->expects($this->exactly(3))
-            ->method('get')
-            ->willReturnOnConsecutiveCalls(
-                [$envelope],
-                [/* idle */],
-                [$envelope, $envelope]
-            );
+        $receiver = new ResettableDummyReceiver([
+            [$envelope],
+            [/* idle */],
+            [$envelope, $envelope],
+        ]);
         $msgCount = 3;
 
         $receiverLocator = $this->createMock(ContainerInterface::class);
@@ -134,8 +131,7 @@ public function testRunWithResetServicesOption(bool $shouldReset)
         $bus = $this->createMock(RoutableMessageBus::class);
         $bus->expects($this->exactly($msgCount))->method('dispatch');
 
-        $servicesResetter = $this->createMock(ServicesResetter::class);
-        $servicesResetter->expects($this->exactly($shouldReset ? $msgCount : 0))->method('reset');
+        $servicesResetter = new ServicesResetter(new \ArrayIterator([$receiver]), ['reset']);
 
         $command = new ConsumeMessagesCommand($bus, $receiverLocator, new EventDispatcher(), null, [], new ResetServicesListener($servicesResetter));
 
@@ -148,6 +144,7 @@ public function testRunWithResetServicesOption(bool $shouldReset)
             '--limit' => $msgCount,
         ], $shouldReset ? [] : ['--no-reset' => null]));
 
+        $this->assertEquals($shouldReset, $receiver->hasBeenReset(), '$receiver->reset() should have been called');
         $tester->assertCommandIsSuccessful();
         $this->assertStringContainsString('[OK] Consuming messages from transports "dummy-receiver"', $tester->getDisplay());
     }

diff --git a/src/Symfony/Component/Messenger/Tests/EventListener/ResetServicesListenerTest.php b/src/Symfony/Component/Messenger/Tests/EventListener/ResetServicesListenerTest.php
@@ -14,6 +14,7 @@
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
 use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
 use Symfony\Component\Messenger\EventListener\ResetServicesListener;
 use Symfony\Component\Messenger\Worker;
 
@@ -38,4 +39,15 @@ public function testResetServices(bool $shouldReset)
         $resetListener = new ResetServicesListener($servicesResetter);
         $resetListener->resetServices($event);
     }
+
+    public function testResetServicesAtStop()
+    {
+        $servicesResetter = $this->createMock(ServicesResetter::class);
+        $servicesResetter->expects($this->once())->method('reset');
+
+        $event = new WorkerStoppedEvent($this->createMock(Worker::class));
+
+        $resetListener = new ResetServicesListener($servicesResetter);
+        $resetListener->resetServicesAtStop($event);
+    }
 }
diff --git a/src/Symfony/Component/Messenger/Tests/WorkerTest.php b/src/Symfony/Component/Messenger/Tests/WorkerTest.php
@@ -14,13 +14,15 @@
 use PHPUnit\Framework\TestCase;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcher;
+use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
 use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
 use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
 use Symfony\Component\Messenger\Event\WorkerRunningEvent;
 use Symfony\Component\Messenger\Event\WorkerStartedEvent;
 use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
+use Symfony\Component\Messenger\EventListener\ResetServicesListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
 use Symfony\Component\Messenger\Exception\RuntimeException;
 use Symfony\Component\Messenger\Handler\Acknowledger;
@@ -103,15 +105,50 @@ public function testWorkerResetsConnectionIfReceiverIsResettable()
     {
         $resettableReceiver = new ResettableDummyReceiver([]);
 
-        $bus = $this->createMock(MessageBusInterface::class);
         $dispatcher = new EventDispatcher();
+        $dispatcher->addSubscriber(new ResetServicesListener(new ServicesResetter(new \ArrayIterator([$resettableReceiver]), ['reset'])));
 
+        $bus = $this->createMock(MessageBusInterface::class);
         $worker = new Worker([$resettableReceiver], $bus, $dispatcher);
         $worker->stop();
         $worker->run();
         $this->assertTrue($resettableReceiver->hasBeenReset());
     }
 
+    public function testWorkerResetsTransportsIfResetServicesListenerIsCalled()
+    {
+        $envelope = new Envelope(new DummyMessage('Hello'));
+        $resettableReceiver = new ResettableDummyReceiver([[$envelope]]);
+
+        $dispatcher = new EventDispatcher();
+        $dispatcher->addSubscriber(new ResetServicesListener(new ServicesResetter(new \ArrayIterator([$resettableReceiver]), ['reset'])));
+        $dispatcher->addListener(WorkerRunningEvent::class, function (WorkerRunningEvent $event) {
+            $event->getWorker()->stop();
+        });
+
+        $bus = $this->createMock(MessageBusInterface::class);
+        $worker = new Worker([$resettableReceiver], $bus, $dispatcher);
+        $worker->run();
+        $this->assertTrue($resettableReceiver->hasBeenReset());
+    }
+
+    public function testWorkerDoesNotResetTransportsIfResetServicesListenerIsNotCalled()
+    {
+        $envelope = new Envelope(new DummyMessage('Hello'));
+        $resettableReceiver = new ResettableDummyReceiver([[$envelope]]);
+
+        $bus = $this->createMock(MessageBusInterface::class);
+
+        $dispatcher = new EventDispatcher();
+        $dispatcher->addListener(WorkerRunningEvent::class, function (WorkerRunningEvent $event) {
+            $event->getWorker()->stop();
+        });
+
+        $worker = new Worker([$resettableReceiver], $bus, $dispatcher);
+        $worker->run();
+        $this->assertFalse($resettableReceiver->hasBeenReset());
+    }
+
     public function testWorkerDoesNotSendNullMessagesToTheBus()
     {
         $receiver = new DummyReceiver([

diff --git a/src/Symfony/Component/Messenger/Worker.php b/src/Symfony/Component/Messenger/Worker.php
@@ -136,7 +136,6 @@ public function run(array $options = []): void
 
         $this->flush(true);
         $this->dispatchEvent(new WorkerStoppedEvent($this));
-        $this->resetReceiverConnections();
     }
 
     private function handleMessage(Envelope $envelope, string $transportName): void
@@ -260,15 +259,6 @@ public function getMetadata(): WorkerMetadata
         return $this->metadata;
     }
 
-    private function resetReceiverConnections(): void
-    {
-        foreach ($this->receivers as $receiver) {
-            if ($receiver instanceof ResetInterface) {
-                $receiver->reset();
-            }
-        }
-    }
-
     private function dispatchEvent(object $event): void
     {
         if (null === $this->eventDispatcher) {