symfony · nicolas-grekas · Sep 29, 2023 · Jul 11, 2023
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",