[Messenger] Fix exiting FailedMessagesRetryCommand

symfony · nicolas-grekas · Sep 29, 2023 · Jul 11, 2023 · Sep 29, 2023 · cd6816b363e806e57f980d7e4b39898985116792
commit cd6816b363e806e57f980d7e4b39898985116792
diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
@@ -2098,6 +2098,16 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->getDefinition('messenger.transport.beanstalkd.factory')->addTag('messenger.transport_factory');
         }
 
+        if ($config['stop_worker_on_signals'] && $this->hasConsole()) {
+            $container->getDefinition('console.command.messenger_consume_messages')
+                ->replaceArgument(8, $config['stop_worker_on_signals']);
+            $container->getDefinition('console.command.messenger_failed_messages_retry')
+                ->replaceArgument(6, $config['stop_worker_on_signals']);
+        }
+
+        if ($this->hasConsole() && $container->hasDefinition('messenger.listener.stop_worker_signals_listener')) {
+            $container->getDefinition('messenger.listener.stop_worker_signals_listener')->clearTag('kernel.event_subscriber');
+        }
         if ($config['stop_worker_on_signals']) {
             $container->getDefinition('messenger.listener.stop_worker_signals_listener')->replaceArgument(0, $config['stop_worker_on_signals']);
         }

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/console.php b/src/Symfony/Bundle/FrameworkBundle/Resources/config/console.php
@@ -163,6 +163,7 @@
                 service('messenger.listener.reset_services')->nullOnInvalid(),
                 [], // Bus names
                 service('messenger.rate_limiter_locator')->nullOnInvalid(),
+                null,
             ])
             ->tag('console.command')
             ->tag('monolog.logger', ['channel' => 'messenger'])
@@ -194,6 +195,7 @@
                 service('event_dispatcher'),
                 service('logger')->nullOnInvalid(),
                 service('messenger.transport.native_php_serializer')->nullOnInvalid(),
+                null,
             ])
             ->tag('console.command')
             ->tag('monolog.logger', ['channel' => 'messenger'])

diff --git a/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php b/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php
@@ -15,6 +15,7 @@
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Command\SignalableCommandInterface;
 use Symfony\Component\Console\Completion\CompletionInput;
 use Symfony\Component\Console\Completion\CompletionSuggestions;
 use Symfony\Component\Console\Exception\InvalidOptionException;
@@ -39,7 +40,7 @@
  * @author Samuel Roze <samuel.roze@gmail.com>
  */
 #[AsCommand(name: 'messenger:consume', description: 'Consume messages')]
-class ConsumeMessagesCommand extends Command
+class ConsumeMessagesCommand extends Command implements SignalableCommandInterface
 {
     private RoutableMessageBus $routableBus;
     private ContainerInterface $receiverLocator;
@@ -49,8 +50,10 @@ class ConsumeMessagesCommand extends Command
     private ?ResetServicesListener $resetServicesListener;
     private array $busIds;
     private ?ContainerInterface $rateLimiterLocator;
+    private ?array $signals;
+    private ?Worker $worker = null;
 
-    public function __construct(RoutableMessageBus $routableBus, ContainerInterface $receiverLocator, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null, array $receiverNames = [], ResetServicesListener $resetServicesListener = null, array $busIds = [], ContainerInterface $rateLimiterLocator = null)
+    public function __construct(RoutableMessageBus $routableBus, ContainerInterface $receiverLocator, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null, array $receiverNames = [], ResetServicesListener $resetServicesListener = null, array $busIds = [], ContainerInterface $rateLimiterLocator = null, array $signals = null)
     {
         $this->routableBus = $routableBus;
         $this->receiverLocator = $receiverLocator;
@@ -60,6 +63,7 @@ public function __construct(RoutableMessageBus $routableBus, ContainerInterface
         $this->resetServicesListener = $resetServicesListener;
         $this->busIds = $busIds;
         $this->rateLimiterLocator = $rateLimiterLocator;
+        $this->signals = $signals;
 
         parent::__construct();
     }
@@ -222,14 +226,19 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $bus = $input->getOption('bus') ? $this->routableBus->getMessageBus($input->getOption('bus')) : $this->routableBus;
 
-        $worker = new Worker($receivers, $bus, $this->eventDispatcher, $this->logger, $rateLimiters);
+        $this->worker = new Worker($receivers, $bus, $this->eventDispatcher, $this->logger, $rateLimiters);
         $options = [
             'sleep' => $input->getOption('sleep') * 1000000,
         ];
         if ($queues = $input->getOption('queues')) {
             $options['queues'] = $queues;
         }
-        $worker->run($options);
+
+        try {
+            $this->worker->run($options);
+        } finally {
+            $this->worker = null;
+        }
 
         return 0;
     }
@@ -247,6 +256,24 @@ public function complete(CompletionInput $input, CompletionSuggestions $suggesti
         }
     }
 
+    public function getSubscribedSignals(): array
+    {
+        return $this->signals ?? [\SIGTERM, \SIGINT];
+    }
+
+    public function handleSignal(int $signal, int|false $previousExitCode = 0): int|false
+    {
+        if (!$this->worker) {
+            return false;
+        }
+
+        $this->logger?->info('Received signal {signal}.', ['signal' => $signal, 'transport_names' => $this->worker->getMetadata()->getTransportNames()]);
+
+        $this->worker->stop();
+
+        return 0;
+    }
+
     private function convertToBytes(string $memoryLimit): int
     {
         $memoryLimit = strtolower($memoryLimit);

diff --git a/src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php b/src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php
@@ -13,6 +13,7 @@
 
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\SignalableCommandInterface;
 use Symfony\Component\Console\Exception\RuntimeException;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
@@ -36,17 +37,20 @@
  * @author Ryan Weaver <ryan@symfonycasts.com>
  */
 #[AsCommand(name: 'messenger:failed:retry', description: 'Retry one or more messages from the failure transport')]
-class FailedMessagesRetryCommand extends AbstractFailedMessagesCommand
+class FailedMessagesRetryCommand extends AbstractFailedMessagesCommand implements SignalableCommandInterface
 {
     private EventDispatcherInterface $eventDispatcher;
     private MessageBusInterface $messageBus;
     private ?LoggerInterface $logger;
+    private ?array $signals;
+    private ?Worker $worker = null;
 
-    public function __construct(?string $globalReceiverName, ServiceProviderInterface $failureTransports, MessageBusInterface $messageBus, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null, PhpSerializer $phpSerializer = null)
+    public function __construct(?string $globalReceiverName, ServiceProviderInterface $failureTransports, MessageBusInterface $messageBus, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null, PhpSerializer $phpSerializer = null, array $signals = null)
     {
         $this->eventDispatcher = $eventDispatcher;
         $this->messageBus = $messageBus;
         $this->logger = $logger;
+        $this->signals = $signals;
 
         parent::__construct($globalReceiverName, $failureTransports, $phpSerializer);
     }
@@ -123,6 +127,24 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         return 0;
     }
 
+    public function getSubscribedSignals(): array
+    {
+        return $this->signals ?? [\SIGTERM, \SIGINT];
+    }
+
+    public function handleSignal(int $signal, int|false $previousExitCode = 0): int|false
+    {
+        if (!$this->worker) {
+            return false;
+        }
+
+        $this->logger?->info('Received signal {signal}.', ['signal' => $signal, 'transport_names' => $this->worker->getMetadata()->getTransportNames()]);
+
+        $this->worker->stop();
+
+        return 0;
+    }
+
     private function runInteractive(string $failureTransportName, SymfonyStyle $io, bool $shouldForce): void
     {
         $receiver = $this->failureTransports->get($failureTransportName);
@@ -187,16 +209,17 @@ private function runWorker(string $failureTransportName, ReceiverInterface $rece
         };
         $this->eventDispatcher->addListener(WorkerMessageReceivedEvent::class, $listener);
 
-        $worker = new Worker(
+        $this->worker = new Worker(
             [$failureTransportName => $receiver],
             $this->messageBus,
             $this->eventDispatcher,
             $this->logger
         );
 
         try {
-            $worker->run();
+            $this->worker->run();
         } finally {
+            $this->worker = null;
             $this->eventDispatcher->removeListener(WorkerMessageReceivedEvent::class, $listener);
         }
 

diff --git a/src/Symfony/Component/Messenger/composer.json b/src/Symfony/Component/Messenger/composer.json
@@ -22,7 +22,7 @@
     },
     "require-dev": {
         "psr/cache": "^1.0|^2.0|^3.0",
-        "symfony/console": "^5.4|^6.0",
+        "symfony/console": "^6.3",
         "symfony/dependency-injection": "^5.4|^6.0",
         "symfony/deprecation-contracts": "^2.5|^3",
         "symfony/event-dispatcher": "^5.4|^6.0",
@@ -37,6 +37,7 @@
         "symfony/validator": "^5.4|^6.0"
     },
     "conflict": {
+        "symfony/console": "<6.3",
         "symfony/event-dispatcher": "<5.4",
         "symfony/event-dispatcher-contracts": "<2.5",
         "symfony/framework-bundle": "<5.4",