Subida del módulo y tema de PrestaShop

vendor/symfony/messenger/Command/AbstractFailedMessagesCommand.php

@@ -0,0 +1,246 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Completion\CompletionInput;
+use Symfony\Component\Console\Completion\CompletionSuggestions;
+use Symfony\Component\Console\Helper\Dumper;
+use Symfony\Component\Console\Question\ChoiceQuestion;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\ErrorHandler\Exception\FlattenException;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\InvalidArgumentException;
+use Symfony\Component\Messenger\Stamp\ErrorDetailsStamp;
+use Symfony\Component\Messenger\Stamp\MessageDecodingFailedStamp;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
+use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp;
+use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
+use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
+use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
+use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
+use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
+use Symfony\Component\VarDumper\Caster\Caster;
+use Symfony\Component\VarDumper\Caster\TraceStub;
+use Symfony\Component\VarDumper\Cloner\ClonerInterface;
+use Symfony\Component\VarDumper\Cloner\Stub;
+use Symfony\Component\VarDumper\Cloner\VarCloner;
+use Symfony\Contracts\Service\ServiceProviderInterface;
+
+/**
+ * @author Ryan Weaver <ryan@symfonycasts.com>
+ *
+ * @internal
+ */
+abstract class AbstractFailedMessagesCommand extends Command
+{
+    protected const DEFAULT_TRANSPORT_OPTION = 'choose';
+
+    protected ServiceProviderInterface $failureTransports;
+    protected ?PhpSerializer $phpSerializer;
+
+    private ?string $globalFailureReceiverName;
+
+    public function __construct(?string $globalFailureReceiverName, ServiceProviderInterface $failureTransports, ?PhpSerializer $phpSerializer = null)
+    {
+        $this->failureTransports = $failureTransports;
+        $this->globalFailureReceiverName = $globalFailureReceiverName;
+        $this->phpSerializer = $phpSerializer;
+
+        parent::__construct();
+    }
+
+    protected function getGlobalFailureReceiverName(): ?string
+    {
+        return $this->globalFailureReceiverName;
+    }
+
+    protected function getMessageId(Envelope $envelope): mixed
+    {
+        /** @var TransportMessageIdStamp $stamp */
+        $stamp = $envelope->last(TransportMessageIdStamp::class);
+
+        return $stamp?->getId();
+    }
+
+    protected function displaySingleMessage(Envelope $envelope, SymfonyStyle $io): void
+    {
+        $io->title('Failed Message Details');
+
+        /** @var SentToFailureTransportStamp|null $sentToFailureTransportStamp */
+        $sentToFailureTransportStamp = $envelope->last(SentToFailureTransportStamp::class);
+        /** @var RedeliveryStamp|null $lastRedeliveryStamp */
+        $lastRedeliveryStamp = $envelope->last(RedeliveryStamp::class);
+        /** @var ErrorDetailsStamp|null $lastErrorDetailsStamp */
+        $lastErrorDetailsStamp = $envelope->last(ErrorDetailsStamp::class);
+        /** @var MessageDecodingFailedStamp|null $lastMessageDecodingFailedStamp */
+        $lastMessageDecodingFailedStamp = $envelope->last(MessageDecodingFailedStamp::class);
+
+        $rows = [
+            ['Class', $envelope->getMessage()::class],
+        ];
+
+        if (null !== $id = $this->getMessageId($envelope)) {
+            $rows[] = ['Message Id', $id];
+        }
+
+        if (null === $sentToFailureTransportStamp) {
+            $io->warning('Message does not appear to have been sent to this transport after failing');
+        } else {
+            $failedAt = '';
+            $errorMessage = '';
+            $errorCode = '';
+            $errorClass = '(unknown)';
+
+            if (null !== $lastRedeliveryStamp) {
+                $failedAt = $lastRedeliveryStamp->getRedeliveredAt()->format('Y-m-d H:i:s');
+            }
+
+            if (null !== $lastErrorDetailsStamp) {
+                $errorMessage = $lastErrorDetailsStamp->getExceptionMessage();
+                $errorCode = $lastErrorDetailsStamp->getExceptionCode();
+                $errorClass = $lastErrorDetailsStamp->getExceptionClass();
+            }
+
+            $rows = array_merge($rows, [
+                ['Failed at', $failedAt],
+                ['Error', $errorMessage],
+                ['Error Code', $errorCode],
+                ['Error Class', $errorClass],
+                ['Transport', $sentToFailureTransportStamp->getOriginalReceiverName()],
+            ]);
+        }
+
+        $io->table([], $rows);
+
+        /** @var RedeliveryStamp[] $redeliveryStamps */
+        $redeliveryStamps = $envelope->all(RedeliveryStamp::class);
+        $io->writeln(' Message history:');
+        foreach ($redeliveryStamps as $redeliveryStamp) {
+            $io->writeln(\sprintf('  * Message failed at <info>%s</info> and was redelivered', $redeliveryStamp->getRedeliveredAt()->format('Y-m-d H:i:s')));
+        }
+        $io->newLine();
+
+        if ($io->isVeryVerbose()) {
+            $io->title('Message:');
+            if (null !== $lastMessageDecodingFailedStamp) {
+                $io->error('The message could not be decoded. See below an APPROXIMATIVE representation of the class.');
+            }
+            $dump = new Dumper($io, null, $this->createCloner());
+            $io->writeln($dump($envelope->getMessage()));
+            $io->title('Exception:');
+            $flattenException = $lastErrorDetailsStamp?->getFlattenException();
+            $io->writeln(null === $flattenException ? '(no data)' : $dump($flattenException));
+        } else {
+            if (null !== $lastMessageDecodingFailedStamp) {
+                $io->error('The message could not be decoded.');
+            }
+            $io->writeln(' Re-run command with <info>-vv</info> to see more message & error details.');
+        }
+    }
+
+    protected function printPendingMessagesMessage(ReceiverInterface $receiver, SymfonyStyle $io): void
+    {
+        if ($receiver instanceof MessageCountAwareInterface) {
+            if (1 === $receiver->getMessageCount()) {
+                $io->writeln('There is <comment>1</comment> message pending in the failure transport.');
+            } else {
+                $io->writeln(\sprintf('There are <comment>%d</comment> messages pending in the failure transport.', $receiver->getMessageCount()));
+            }
+        }
+    }
+
+    protected function getReceiver(?string $name = null): ReceiverInterface
+    {
+        if (null === $name ??= $this->globalFailureReceiverName) {
+            throw new InvalidArgumentException(\sprintf('No default failure transport is defined. Available transports are: "%s".', implode('", "', array_keys($this->failureTransports->getProvidedServices()))));
+        }
+
+        if (!$this->failureTransports->has($name)) {
+            throw new InvalidArgumentException(\sprintf('The "%s" failure transport was not found. Available transports are: "%s".', $name, implode('", "', array_keys($this->failureTransports->getProvidedServices()))));
+        }
+
+        return $this->failureTransports->get($name);
+    }
+
+    private function createCloner(): ?ClonerInterface
+    {
+        if (!class_exists(VarCloner::class)) {
+            return null;
+        }
+
+        $cloner = new VarCloner();
+        $cloner->addCasters([FlattenException::class => function (FlattenException $flattenException, array $a, Stub $stub): array {
+            $stub->class = $flattenException->getClass();
+
+            return [
+                Caster::PREFIX_VIRTUAL.'message' => $flattenException->getMessage(),
+                Caster::PREFIX_VIRTUAL.'code' => $flattenException->getCode(),
+                Caster::PREFIX_VIRTUAL.'file' => $flattenException->getFile(),
+                Caster::PREFIX_VIRTUAL.'line' => $flattenException->getLine(),
+                Caster::PREFIX_VIRTUAL.'trace' => new TraceStub($flattenException->getTrace()),
+            ];
+        }]);
+
+        return $cloner;
+    }
+
+    protected function printWarningAvailableFailureTransports(SymfonyStyle $io, ?string $failureTransportName): void
+    {
+        $failureTransports = array_keys($this->failureTransports->getProvidedServices());
+        $failureTransportsCount = \count($failureTransports);
+        if ($failureTransportsCount > 1) {
+            $io->writeln([
+                \sprintf('> Loading messages from the <comment>global</comment> failure transport <comment>%s</comment>.', $failureTransportName),
+                '> To use a different failure transport, pass <comment>--transport=</comment>.',
+                \sprintf('> Available failure transports are: <comment>%s</comment>', implode(', ', $failureTransports)),
+                "\n",
+            ]);
+        }
+    }
+
+    protected function interactiveChooseFailureTransport(SymfonyStyle $io): string
+    {
+        $failedTransports = array_keys($this->failureTransports->getProvidedServices());
+        $question = new ChoiceQuestion('Select failed transport:', $failedTransports, 0);
+        $question->setMultiselect(false);
+
+        return $io->askQuestion($question);
+    }
+
+    public function complete(CompletionInput $input, CompletionSuggestions $suggestions): void
+    {
+        if ($input->mustSuggestOptionValuesFor('transport')) {
+            $suggestions->suggestValues(array_keys($this->failureTransports->getProvidedServices()));
+
+            return;
+        }
+
+        if ($input->mustSuggestArgumentValuesFor('id')) {
+            $transport = $input->getOption('transport');
+            $transport = self::DEFAULT_TRANSPORT_OPTION === $transport ? $this->getGlobalFailureReceiverName() : $transport;
+            $receiver = $this->getReceiver($transport);
+
+            if (!$receiver instanceof ListableReceiverInterface) {
+                return;
+            }
+
+            $ids = [];
+            foreach ($receiver->all(50) as $envelope) {
+                $ids[] = $this->getMessageId($envelope);
+            }
+            $suggestions->suggestValues($ids);
+
+            return;
+        }
+    }
+}

vendor/symfony/messenger/Command/ConsumeMessagesCommand.php

@@ -0,0 +1,307 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Psr\Container\ContainerInterface;
+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;
+use Symfony\Component\Console\Exception\RuntimeException;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\ConsoleOutputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Question\ChoiceQuestion;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\Messenger\EventListener\ResetServicesListener;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnFailureLimitListener;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnMemoryLimitListener;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnTimeLimitListener;
+use Symfony\Component\Messenger\RoutableMessageBus;
+use Symfony\Component\Messenger\Worker;
+
+/**
+ * @author Samuel Roze <samuel.roze@gmail.com>
+ */
+#[AsCommand(name: 'messenger:consume', description: 'Consume messages')]
+class ConsumeMessagesCommand extends Command implements SignalableCommandInterface
+{
+    private RoutableMessageBus $routableBus;
+    private ContainerInterface $receiverLocator;
+    private EventDispatcherInterface $eventDispatcher;
+    private ?LoggerInterface $logger;
+    private array $receiverNames;
+    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, ?array $signals = null)
+    {
+        $this->routableBus = $routableBus;
+        $this->receiverLocator = $receiverLocator;
+        $this->eventDispatcher = $eventDispatcher;
+        $this->logger = $logger;
+        $this->receiverNames = $receiverNames;
+        $this->resetServicesListener = $resetServicesListener;
+        $this->busIds = $busIds;
+        $this->rateLimiterLocator = $rateLimiterLocator;
+        $this->signals = $signals;
+
+        parent::__construct();
+    }
+
+    protected function configure(): void
+    {
+        $defaultReceiverName = 1 === \count($this->receiverNames) ? current($this->receiverNames) : null;
+
+        $this
+            ->setDefinition([
+                new InputArgument('receivers', InputArgument::IS_ARRAY, 'Names of the receivers/transports to consume in order of priority', $defaultReceiverName ? [$defaultReceiverName] : []),
+                new InputOption('limit', 'l', InputOption::VALUE_REQUIRED, 'Limit the number of received messages'),
+                new InputOption('failure-limit', 'f', InputOption::VALUE_REQUIRED, 'The number of failed messages the worker can consume'),
+                new InputOption('memory-limit', 'm', InputOption::VALUE_REQUIRED, 'The memory limit the worker can consume'),
+                new InputOption('time-limit', 't', InputOption::VALUE_REQUIRED, 'The time limit in seconds the worker can handle new messages'),
+                new InputOption('sleep', null, InputOption::VALUE_REQUIRED, 'Seconds to sleep before asking for new messages after no messages were found', 1),
+                new InputOption('bus', 'b', InputOption::VALUE_REQUIRED, 'Name of the bus to which received messages should be dispatched (if not passed, bus is determined automatically)'),
+                new InputOption('queues', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Limit receivers to only consume from the specified queues'),
+                new InputOption('no-reset', null, InputOption::VALUE_NONE, 'Do not reset container services after each message'),
+            ])
+            ->setHelp(<<<'EOF'
+The <info>%command.name%</info> command consumes messages and dispatches them to the message bus.
+
+    <info>php %command.full_name% <receiver-name></info>
+
+To receive from multiple transports, pass each name:
+
+    <info>php %command.full_name% receiver1 receiver2</info>
+
+Use the --limit option to limit the number of messages received:
+
+    <info>php %command.full_name% <receiver-name> --limit=10</info>
+
+Use the --failure-limit option to stop the worker when the given number of failed messages is reached:
+
+    <info>php %command.full_name% <receiver-name> --failure-limit=2</info>
+
+Use the --memory-limit option to stop the worker if it exceeds a given memory usage limit. You can use shorthand byte values [K, M or G]:
+
+    <info>php %command.full_name% <receiver-name> --memory-limit=128M</info>
+
+Use the --time-limit option to stop the worker when the given time limit (in seconds) is reached.
+If a message is being handled, the worker will stop after the processing is finished:
+
+    <info>php %command.full_name% <receiver-name> --time-limit=3600</info>
+
+Use the --bus option to specify the message bus to dispatch received messages
+to instead of trying to determine it automatically. This is required if the
+messages didn't originate from Messenger:
+
+    <info>php %command.full_name% <receiver-name> --bus=event_bus</info>
+
+Use the --queues option to limit a receiver to only certain queues (only supported by some receivers):
+
+    <info>php %command.full_name% <receiver-name> --queues=fasttrack</info>
+
+Use the --no-reset option to prevent services resetting after each message (may lead to leaking services' state between messages):
+
+    <info>php %command.full_name% <receiver-name> --no-reset</info>
+EOF
+            )
+        ;
+    }
+
+    /**
+     * @return void
+     */
+    protected function interact(InputInterface $input, OutputInterface $output)
+    {
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+
+        if ($this->receiverNames && !$input->getArgument('receivers')) {
+            if (1 === \count($this->receiverNames)) {
+                $input->setArgument('receivers', $this->receiverNames);
+
+                return;
+            }
+
+            $io->block('Which transports/receivers do you want to consume?', null, 'fg=white;bg=blue', ' ', true);
+
+            $io->writeln('Choose which receivers you want to consume messages from in order of priority.');
+            if (\count($this->receiverNames) > 1) {
+                $io->writeln(\sprintf('Hint: to consume from multiple, use a list of their names, e.g. <comment>%s</comment>', implode(', ', $this->receiverNames)));
+            }
+
+            $question = new ChoiceQuestion('Select receivers to consume:', $this->receiverNames, 0);
+            $question->setMultiselect(true);
+
+            $input->setArgument('receivers', $io->askQuestion($question));
+        }
+
+        if (!$input->getArgument('receivers')) {
+            throw new RuntimeException('Please pass at least one receiver.');
+        }
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $receivers = [];
+        $rateLimiters = [];
+        foreach ($receiverNames = $input->getArgument('receivers') as $receiverName) {
+            if (!$this->receiverLocator->has($receiverName)) {
+                $message = \sprintf('The receiver "%s" does not exist.', $receiverName);
+                if ($this->receiverNames) {
+                    $message .= \sprintf(' Valid receivers are: %s.', implode(', ', $this->receiverNames));
+                }
+
+                throw new RuntimeException($message);
+            }
+
+            $receivers[$receiverName] = $this->receiverLocator->get($receiverName);
+            if ($this->rateLimiterLocator?->has($receiverName)) {
+                $rateLimiters[$receiverName] = $this->rateLimiterLocator->get($receiverName);
+            }
+        }
+
+        if (null !== $this->resetServicesListener && !$input->getOption('no-reset')) {
+            $this->eventDispatcher->addSubscriber($this->resetServicesListener);
+        }
+
+        $stopsWhen = [];
+        if (null !== $limit = $input->getOption('limit')) {
+            if (!is_numeric($limit) || 0 >= $limit) {
+                throw new InvalidOptionException(\sprintf('Option "limit" must be a positive integer, "%s" passed.', $limit));
+            }
+
+            $stopsWhen[] = "processed {$limit} messages";
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener($limit, $this->logger));
+        }
+
+        if ($failureLimit = $input->getOption('failure-limit')) {
+            $stopsWhen[] = "reached {$failureLimit} failed messages";
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnFailureLimitListener($failureLimit, $this->logger));
+        }
+
+        if ($memoryLimit = $input->getOption('memory-limit')) {
+            $stopsWhen[] = "exceeded {$memoryLimit} of memory";
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnMemoryLimitListener($this->convertToBytes($memoryLimit), $this->logger));
+        }
+
+        if (null !== $timeLimit = $input->getOption('time-limit')) {
+            if (!is_numeric($timeLimit) || 0 >= $timeLimit) {
+                throw new InvalidOptionException(\sprintf('Option "time-limit" must be a positive integer, "%s" passed.', $timeLimit));
+            }
+
+            $stopsWhen[] = "been running for {$timeLimit}s";
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnTimeLimitListener($timeLimit, $this->logger));
+        }
+
+        $stopsWhen[] = 'received a stop signal via the messenger:stop-workers command';
+
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+        $io->success(\sprintf('Consuming messages from transport%s "%s".', \count($receivers) > 1 ? 's' : '', implode(', ', $receiverNames)));
+
+        if ($stopsWhen) {
+            $last = array_pop($stopsWhen);
+            $stopsWhen = ($stopsWhen ? implode(', ', $stopsWhen).' or ' : '').$last;
+            $io->comment("The worker will automatically exit once it has {$stopsWhen}.");
+        }
+
+        $io->comment('Quit the worker with CONTROL-C.');
+
+        if (OutputInterface::VERBOSITY_VERBOSE > $output->getVerbosity()) {
+            $io->comment('Re-run the command with a -vv option to see logs about consumed messages.');
+        }
+
+        $bus = $input->getOption('bus') ? $this->routableBus->getMessageBus($input->getOption('bus')) : $this->routableBus;
+
+        $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;
+        }
+
+        try {
+            $this->worker->run($options);
+        } finally {
+            $this->worker = null;
+        }
+
+        return 0;
+    }
+
+    public function complete(CompletionInput $input, CompletionSuggestions $suggestions): void
+    {
+        if ($input->mustSuggestArgumentValuesFor('receivers')) {
+            $suggestions->suggestValues(array_diff($this->receiverNames, array_diff($input->getArgument('receivers'), [$input->getCompletionValue()])));
+
+            return;
+        }
+
+        if ($input->mustSuggestOptionValuesFor('bus')) {
+            $suggestions->suggestValues($this->busIds);
+        }
+    }
+
+    public function getSubscribedSignals(): array
+    {
+        return $this->signals ?? (\extension_loaded('pcntl') ? [\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 false;
+    }
+
+    private function convertToBytes(string $memoryLimit): int
+    {
+        $memoryLimit = strtolower($memoryLimit);
+        $max = ltrim($memoryLimit, '+');
+        if (str_starts_with($max, '0x')) {
+            $max = \intval($max, 16);
+        } elseif (str_starts_with($max, '0')) {
+            $max = \intval($max, 8);
+        } else {
+            $max = (float) $max;
+        }
+
+        switch (substr(rtrim($memoryLimit, 'b'), -1)) {
+            case 't': $max *= 1024;
+                // no break
+            case 'g': $max *= 1024;
+                // no break
+            case 'm': $max *= 1024;
+                // no break
+            case 'k': $max *= 1024;
+        }
+
+        return (int) $max;
+    }
+}

vendor/symfony/messenger/Command/DebugCommand.php

@@ -0,0 +1,145 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Completion\CompletionInput;
+use Symfony\Component\Console\Completion\CompletionSuggestions;
+use Symfony\Component\Console\Exception\RuntimeException;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+
+/**
+ * A console command to debug Messenger information.
+ *
+ * @author Roland Franssen <franssen.roland@gmail.com>
+ */
+#[AsCommand(name: 'debug:messenger', description: 'List messages you can dispatch using the message buses')]
+class DebugCommand extends Command
+{
+    private array $mapping;
+
+    public function __construct(array $mapping)
+    {
+        $this->mapping = $mapping;
+
+        parent::__construct();
+    }
+
+    /**
+     * @return void
+     */
+    protected function configure()
+    {
+        $this
+            ->addArgument('bus', InputArgument::OPTIONAL, \sprintf('The bus id (one of "%s")', implode('", "', array_keys($this->mapping))))
+            ->setHelp(<<<'EOF'
+The <info>%command.name%</info> command displays all messages that can be
+dispatched using the message buses:
+
+  <info>php %command.full_name%</info>
+
+Or for a specific bus only:
+
+  <info>php %command.full_name% command_bus</info>
+
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output);
+        $io->title('Messenger');
+
+        $mapping = $this->mapping;
+        if ($bus = $input->getArgument('bus')) {
+            if (!isset($mapping[$bus])) {
+                throw new RuntimeException(\sprintf('Bus "%s" does not exist. Known buses are "%s".', $bus, implode('", "', array_keys($this->mapping))));
+            }
+            $mapping = [$bus => $mapping[$bus]];
+        }
+
+        foreach ($mapping as $bus => $handlersByMessage) {
+            $io->section($bus);
+
+            $tableRows = [];
+            foreach ($handlersByMessage as $message => $handlers) {
+                if ($description = self::getClassDescription($message)) {
+                    $tableRows[] = [\sprintf('<comment>%s</>', $description)];
+                }
+
+                $tableRows[] = [\sprintf('<fg=cyan>%s</fg=cyan>', $message)];
+                foreach ($handlers as $handler) {
+                    $tableRows[] = [
+                        \sprintf('    handled by <info>%s</>', $handler[0]).$this->formatConditions($handler[1]),
+                    ];
+                    if ($handlerDescription = self::getClassDescription($handler[0])) {
+                        $tableRows[] = [\sprintf('               <comment>%s</>', $handlerDescription)];
+                    }
+                }
+                $tableRows[] = [''];
+            }
+
+            if ($tableRows) {
+                $io->text('The following messages can be dispatched:');
+                $io->newLine();
+                $io->table([], $tableRows);
+            } else {
+                $io->warning(\sprintf('No handled message found in bus "%s".', $bus));
+            }
+        }
+
+        return 0;
+    }
+
+    private function formatConditions(array $options): string
+    {
+        if (!$options) {
+            return '';
+        }
+
+        $optionsMapping = [];
+        foreach ($options as $key => $value) {
+            $optionsMapping[] = $key.'='.$value;
+        }
+
+        return ' (when '.implode(', ', $optionsMapping).')';
+    }
+
+    private static function getClassDescription(string $class): string
+    {
+        try {
+            $r = new \ReflectionClass($class);
+
+            if ($docComment = $r->getDocComment()) {
+                $docComment = preg_split('#\n\s*\*\s*[\n@]#', substr($docComment, 3, -2), 2)[0];
+
+                return trim(preg_replace('#\s*\n\s*\*\s*#', ' ', $docComment));
+            }
+        } catch (\ReflectionException) {
+        }
+
+        return '';
+    }
+
+    public function complete(CompletionInput $input, CompletionSuggestions $suggestions): void
+    {
+        if ($input->mustSuggestArgumentValuesFor('bus')) {
+            $suggestions->suggestValues(array_keys($this->mapping));
+        }
+    }
+}

vendor/symfony/messenger/Command/FailedMessagesRemoveCommand.php

@@ -0,0 +1,148 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Exception\RuntimeException;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\ConsoleOutputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
+use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
+
+/**
+ * @author Ryan Weaver <ryan@symfonycasts.com>
+ */
+#[AsCommand(name: 'messenger:failed:remove', description: 'Remove given messages from the failure transport')]
+class FailedMessagesRemoveCommand extends AbstractFailedMessagesCommand
+{
+    protected function configure(): void
+    {
+        $this
+            ->setDefinition([
+                new InputArgument('id', InputArgument::OPTIONAL | InputArgument::IS_ARRAY, 'Specific message id(s) to remove'),
+                new InputOption('all', null, InputOption::VALUE_NONE, 'Remove all failed messages from the transport'),
+                new InputOption('force', null, InputOption::VALUE_NONE, 'Force the operation without confirmation'),
+                new InputOption('transport', null, InputOption::VALUE_REQUIRED, 'Use a specific failure transport', self::DEFAULT_TRANSPORT_OPTION),
+                new InputOption('show-messages', null, InputOption::VALUE_NONE, 'Display messages before removing it (if multiple ids are given)'),
+            ])
+            ->setHelp(<<<'EOF'
+The <info>%command.name%</info> removes given messages that are pending in the failure transport.
+
+    <info>php %command.full_name% {id1} [{id2} ...]</info>
+
+The specific ids can be found via the messenger:failed:show command.
+
+You can remove all failed messages from the failure transport by using the "--all" option:
+
+    <info>php %command.full_name% --all</info>
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+
+        $failureTransportName = $input->getOption('transport');
+        if (self::DEFAULT_TRANSPORT_OPTION === $failureTransportName) {
+            $failureTransportName = $this->getGlobalFailureReceiverName();
+        }
+
+        $receiver = $this->getReceiver($failureTransportName);
+
+        $shouldForce = $input->getOption('force');
+        $ids = (array) $input->getArgument('id');
+        $shouldDeleteAllMessages = $input->getOption('all');
+
+        $idsCount = \count($ids);
+        if (!$shouldDeleteAllMessages && !$idsCount) {
+            throw new RuntimeException('Please specify at least one message id. If you want to remove all failed messages, use the "--all" option.');
+        } elseif ($shouldDeleteAllMessages && $idsCount) {
+            throw new RuntimeException('You cannot specify message ids when using the "--all" option.');
+        }
+
+        $shouldDisplayMessages = $input->getOption('show-messages') || 1 === $idsCount;
+
+        if (!$receiver instanceof ListableReceiverInterface) {
+            throw new RuntimeException(\sprintf('The "%s" receiver does not support removing specific messages.', $failureTransportName));
+        }
+
+        if ($shouldDeleteAllMessages) {
+            $this->removeAllMessages($receiver, $io, $shouldForce, $shouldDisplayMessages);
+        } else {
+            $this->removeMessagesById($ids, $receiver, $io, $shouldForce, $shouldDisplayMessages);
+        }
+
+        return 0;
+    }
+
+    private function removeMessagesById(array $ids, ListableReceiverInterface $receiver, SymfonyStyle $io, bool $shouldForce, bool $shouldDisplayMessages): void
+    {
+        foreach ($ids as $id) {
+            $this->phpSerializer?->acceptPhpIncompleteClass();
+            try {
+                $envelope = $receiver->find($id);
+            } finally {
+                $this->phpSerializer?->rejectPhpIncompleteClass();
+            }
+
+            if (null === $envelope) {
+                $io->error(\sprintf('The message with id "%s" was not found.', $id));
+                continue;
+            }
+
+            if ($shouldDisplayMessages) {
+                $this->displaySingleMessage($envelope, $io);
+            }
+
+            if ($shouldForce || $io->confirm('Do you want to permanently remove this message?', false)) {
+                $receiver->reject($envelope);
+
+                $io->success(\sprintf('Message with id %s removed.', $id));
+            } else {
+                $io->note(\sprintf('Message with id %s not removed.', $id));
+            }
+        }
+    }
+
+    private function removeAllMessages(ListableReceiverInterface $receiver, SymfonyStyle $io, bool $shouldForce, bool $shouldDisplayMessages): void
+    {
+        if (!$shouldForce) {
+            if ($receiver instanceof MessageCountAwareInterface) {
+                $question = \sprintf('Do you want to permanently remove all (%d) messages?', $receiver->getMessageCount());
+            } else {
+                $question = 'Do you want to permanently remove all failed messages?';
+            }
+
+            if (!$io->confirm($question, false)) {
+                return;
+            }
+        }
+
+        $count = 0;
+        foreach ($receiver->all() as $envelope) {
+            if ($shouldDisplayMessages) {
+                $this->displaySingleMessage($envelope, $io);
+            }
+
+            $receiver->reject($envelope);
+            ++$count;
+        }
+
+        $io->note(\sprintf('%d messages were removed.', $count));
+    }
+}

vendor/symfony/messenger/Command/FailedMessagesRetryCommand.php

@@ -0,0 +1,281 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+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;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\ConsoleOutputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\Stamp\MessageDecodingFailedStamp;
+use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
+use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
+use Symfony\Component\Messenger\Transport\Receiver\SingleMessageReceiver;
+use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
+use Symfony\Component\Messenger\Worker;
+use Symfony\Contracts\Service\ServiceProviderInterface;
+
+/**
+ * @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 implements SignalableCommandInterface
+{
+    private EventDispatcherInterface $eventDispatcher;
+    private MessageBusInterface $messageBus;
+    private ?LoggerInterface $logger;
+    private ?array $signals;
+    private bool $shouldStop = false;
+    private bool $forceExit = false;
+    private ?Worker $worker = 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);
+    }
+
+    protected function configure(): void
+    {
+        $this
+            ->setDefinition([
+                new InputArgument('id', InputArgument::IS_ARRAY, 'Specific message id(s) to retry'),
+                new InputOption('force', null, InputOption::VALUE_NONE, 'Force action without confirmation'),
+                new InputOption('transport', null, InputOption::VALUE_REQUIRED, 'Use a specific failure transport', self::DEFAULT_TRANSPORT_OPTION),
+            ])
+            ->setHelp(<<<'EOF'
+The <info>%command.name%</info> retries message in the failure transport.
+
+    <info>php %command.full_name%</info>
+
+The command will interactively ask if each message should be retried
+or discarded.
+
+Some transports support retrying a specific message id, which comes
+from the <info>messenger:failed:show</info> command.
+
+    <info>php %command.full_name% {id}</info>
+
+Or pass multiple ids at once to process multiple messages:
+
+<info>php %command.full_name% {id1} {id2} {id3}</info>
+
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $this->eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1));
+
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+        $io->comment('Quit this command with CONTROL-C.');
+        if (!$output->isVeryVerbose()) {
+            $io->comment('Re-run the command with a -vv option to see logs about consumed messages.');
+        }
+
+        $failureTransportName = $input->getOption('transport');
+        if (self::DEFAULT_TRANSPORT_OPTION === $failureTransportName) {
+            $this->printWarningAvailableFailureTransports($io, $this->getGlobalFailureReceiverName());
+        }
+        if ('' === $failureTransportName || null === $failureTransportName) {
+            $failureTransportName = $this->interactiveChooseFailureTransport($io);
+        }
+        $failureTransportName = self::DEFAULT_TRANSPORT_OPTION === $failureTransportName ? $this->getGlobalFailureReceiverName() : $failureTransportName;
+
+        $receiver = $this->getReceiver($failureTransportName);
+        $this->printPendingMessagesMessage($receiver, $io);
+
+        $io->writeln(\sprintf('To retry all the messages, run <comment>messenger:consume %s</comment>', $failureTransportName));
+
+        $shouldForce = $input->getOption('force');
+        $ids = $input->getArgument('id');
+        if (0 === \count($ids)) {
+            if (!$input->isInteractive()) {
+                throw new RuntimeException('Message id must be passed when in non-interactive mode.');
+            }
+
+            $this->runInteractive($failureTransportName, $io, $shouldForce);
+
+            return 0;
+        }
+
+        $this->retrySpecificIds($failureTransportName, $ids, $io, $shouldForce);
+
+        if (!$this->shouldStop) {
+            $io->success('All done!');
+        }
+
+        return 0;
+    }
+
+    public function getSubscribedSignals(): array
+    {
+        return $this->signals ?? (\extension_loaded('pcntl') ? [\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();
+        $this->shouldStop = true;
+
+        return $this->forceExit ? 0 : false;
+    }
+
+    private function runInteractive(string $failureTransportName, SymfonyStyle $io, bool $shouldForce): void
+    {
+        $receiver = $this->failureTransports->get($failureTransportName);
+        $count = 0;
+        if ($receiver instanceof ListableReceiverInterface) {
+            // for listable receivers, find the messages one-by-one
+            // this avoids using get(), which for some less-robust
+            // transports (like Doctrine), will cause the message
+            // to be temporarily "acked", even if the user aborts
+            // handling the message
+            while (!$this->shouldStop) {
+                $envelopes = [];
+                $this->phpSerializer?->acceptPhpIncompleteClass();
+                try {
+                    foreach ($receiver->all(1) as $envelope) {
+                        ++$count;
+                        $envelopes[] = $envelope;
+                    }
+                } finally {
+                    $this->phpSerializer?->rejectPhpIncompleteClass();
+                }
+
+                // break the loop if all messages are consumed
+                if (0 === \count($envelopes)) {
+                    break;
+                }
+
+                $this->retrySpecificEnvelopes($envelopes, $failureTransportName, $io, $shouldForce);
+            }
+        } else {
+            // get() and ask messages one-by-one
+            $count = $this->runWorker($failureTransportName, $receiver, $io, $shouldForce);
+        }
+
+        // avoid success message if nothing was processed
+        if (1 <= $count && !$this->shouldStop) {
+            $io->success('All failed messages have been handled or removed!');
+        }
+    }
+
+    private function runWorker(string $failureTransportName, ReceiverInterface $receiver, SymfonyStyle $io, bool $shouldForce): int
+    {
+        $count = 0;
+        $listener = function (WorkerMessageReceivedEvent $messageReceivedEvent) use ($io, $receiver, $shouldForce, &$count) {
+            ++$count;
+            $envelope = $messageReceivedEvent->getEnvelope();
+
+            $this->displaySingleMessage($envelope, $io);
+
+            if ($envelope->last(MessageDecodingFailedStamp::class)) {
+                throw new \RuntimeException(\sprintf('The message with id "%s" could not decoded, it can only be shown or removed.', $this->getMessageId($envelope) ?? '?'));
+            }
+
+            $this->forceExit = true;
+            try {
+                $shouldHandle = $shouldForce || 'retry' === $io->choice('Please select an action', ['retry', 'delete'], 'retry');
+            } finally {
+                $this->forceExit = false;
+            }
+
+            if ($shouldHandle) {
+                return;
+            }
+
+            $messageReceivedEvent->shouldHandle(false);
+            $receiver->reject($envelope);
+        };
+        $this->eventDispatcher->addListener(WorkerMessageReceivedEvent::class, $listener);
+
+        $this->worker = new Worker(
+            [$failureTransportName => $receiver],
+            $this->messageBus,
+            $this->eventDispatcher,
+            $this->logger
+        );
+
+        try {
+            $this->worker->run();
+        } finally {
+            $this->worker = null;
+            $this->eventDispatcher->removeListener(WorkerMessageReceivedEvent::class, $listener);
+        }
+
+        return $count;
+    }
+
+    private function retrySpecificIds(string $failureTransportName, array $ids, SymfonyStyle $io, bool $shouldForce): void
+    {
+        $receiver = $this->getReceiver($failureTransportName);
+
+        if (!$receiver instanceof ListableReceiverInterface) {
+            throw new RuntimeException(\sprintf('The "%s" receiver does not support retrying messages by id.', $failureTransportName));
+        }
+
+        foreach ($ids as $id) {
+            $this->phpSerializer?->acceptPhpIncompleteClass();
+            try {
+                $envelope = $receiver->find($id);
+            } finally {
+                $this->phpSerializer?->rejectPhpIncompleteClass();
+            }
+            if (null === $envelope) {
+                throw new RuntimeException(\sprintf('The message "%s" was not found.', $id));
+            }
+
+            $singleReceiver = new SingleMessageReceiver($receiver, $envelope);
+            $this->runWorker($failureTransportName, $singleReceiver, $io, $shouldForce);
+
+            if ($this->shouldStop) {
+                break;
+            }
+        }
+    }
+
+    private function retrySpecificEnvelopes(array $envelopes, string $failureTransportName, SymfonyStyle $io, bool $shouldForce): void
+    {
+        $receiver = $this->getReceiver($failureTransportName);
+
+        foreach ($envelopes as $envelope) {
+            $singleReceiver = new SingleMessageReceiver($receiver, $envelope);
+            $this->runWorker($failureTransportName, $singleReceiver, $io, $shouldForce);
+
+            if ($this->shouldStop) {
+                break;
+            }
+        }
+    }
+}

vendor/symfony/messenger/Command/FailedMessagesShowCommand.php

@@ -0,0 +1,197 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Exception\RuntimeException;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\ConsoleOutputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Messenger\Stamp\ErrorDetailsStamp;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
+use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
+
+/**
+ * @author Ryan Weaver <ryan@symfonycasts.com>
+ */
+#[AsCommand(name: 'messenger:failed:show', description: 'Show one or more messages from the failure transport')]
+class FailedMessagesShowCommand extends AbstractFailedMessagesCommand
+{
+    protected function configure(): void
+    {
+        $this
+            ->setDefinition([
+                new InputArgument('id', InputArgument::OPTIONAL, 'Specific message id to show'),
+                new InputOption('max', null, InputOption::VALUE_REQUIRED, 'Maximum number of messages to list', 50),
+                new InputOption('transport', null, InputOption::VALUE_REQUIRED, 'Use a specific failure transport', self::DEFAULT_TRANSPORT_OPTION),
+                new InputOption('stats', null, InputOption::VALUE_NONE, 'Display the message count by class'),
+                new InputOption('class-filter', null, InputOption::VALUE_REQUIRED, 'Filter by a specific class name'),
+            ])
+            ->setHelp(<<<'EOF'
+The <info>%command.name%</info> shows message that are pending in the failure transport.
+
+    <info>php %command.full_name%</info>
+
+Or look at a specific message by its id:
+
+    <info>php %command.full_name% {id}</info>
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+
+        $failureTransportName = $input->getOption('transport');
+        if (self::DEFAULT_TRANSPORT_OPTION === $failureTransportName) {
+            $this->printWarningAvailableFailureTransports($io, $this->getGlobalFailureReceiverName());
+        }
+        if ('' === $failureTransportName || null === $failureTransportName) {
+            $failureTransportName = $this->interactiveChooseFailureTransport($io);
+        }
+        $failureTransportName = self::DEFAULT_TRANSPORT_OPTION === $failureTransportName ? $this->getGlobalFailureReceiverName() : $failureTransportName;
+
+        $receiver = $this->getReceiver($failureTransportName);
+
+        $this->printPendingMessagesMessage($receiver, $io);
+
+        if (!$receiver instanceof ListableReceiverInterface) {
+            throw new RuntimeException(\sprintf('The "%s" receiver does not support listing or showing specific messages.', $failureTransportName));
+        }
+
+        if ($input->getOption('stats')) {
+            $this->listMessagesPerClass($failureTransportName, $io, $input->getOption('max'));
+        } elseif (null === $id = $input->getArgument('id')) {
+            $this->listMessages($failureTransportName, $io, $input->getOption('max'), $input->getOption('class-filter'));
+        } else {
+            $this->showMessage($failureTransportName, $id, $io);
+        }
+
+        return 0;
+    }
+
+    private function listMessages(?string $failedTransportName, SymfonyStyle $io, int $max, ?string $classFilter = null): void
+    {
+        /** @var ListableReceiverInterface $receiver */
+        $receiver = $this->getReceiver($failedTransportName);
+        $envelopes = $receiver->all($max);
+
+        $rows = [];
+
+        if ($classFilter) {
+            $io->comment(\sprintf('Displaying only \'%s\' messages', $classFilter));
+        }
+
+        $this->phpSerializer?->acceptPhpIncompleteClass();
+        try {
+            foreach ($envelopes as $envelope) {
+                $currentClassName = $envelope->getMessage()::class;
+
+                if ($classFilter && $classFilter !== $currentClassName) {
+                    continue;
+                }
+
+                /** @var RedeliveryStamp|null $lastRedeliveryStamp */
+                $lastRedeliveryStamp = $envelope->last(RedeliveryStamp::class);
+                /** @var ErrorDetailsStamp|null $lastErrorDetailsStamp */
+                $lastErrorDetailsStamp = $envelope->last(ErrorDetailsStamp::class);
+
+                $rows[] = [
+                    $this->getMessageId($envelope),
+                    $currentClassName,
+                    null === $lastRedeliveryStamp ? '' : $lastRedeliveryStamp->getRedeliveredAt()->format('Y-m-d H:i:s'),
+                    $lastErrorDetailsStamp?->getExceptionMessage() ?? '',
+                ];
+            }
+        } finally {
+            $this->phpSerializer?->rejectPhpIncompleteClass();
+        }
+
+        $rowsCount = \count($rows);
+
+        if (0 === $rowsCount) {
+            $io->success('No failed messages were found.');
+
+            return;
+        }
+
+        $io->table(['Id', 'Class', 'Failed at', 'Error'], $rows);
+
+        if ($rowsCount === $max) {
+            $io->comment(\sprintf('Showing first %d messages.', $max));
+        } elseif ($classFilter) {
+            $io->comment(\sprintf('Showing %d message(s).', $rowsCount));
+        }
+
+        $io->comment(\sprintf('Run <comment>messenger:failed:show {id} --transport=%s -vv</comment> to see message details.', $failedTransportName));
+    }
+
+    private function listMessagesPerClass(?string $failedTransportName, SymfonyStyle $io, int $max): void
+    {
+        /** @var ListableReceiverInterface $receiver */
+        $receiver = $this->getReceiver($failedTransportName);
+        $envelopes = $receiver->all($max);
+
+        $countPerClass = [];
+
+        $this->phpSerializer?->acceptPhpIncompleteClass();
+        try {
+            foreach ($envelopes as $envelope) {
+                $c = $envelope->getMessage()::class;
+
+                if (!isset($countPerClass[$c])) {
+                    $countPerClass[$c] = [$c, 0];
+                }
+
+                ++$countPerClass[$c][1];
+            }
+        } finally {
+            $this->phpSerializer?->rejectPhpIncompleteClass();
+        }
+
+        if (0 === \count($countPerClass)) {
+            $io->success('No failed messages were found.');
+
+            return;
+        }
+
+        $io->table(['Class', 'Count'], $countPerClass);
+    }
+
+    private function showMessage(?string $failedTransportName, string $id, SymfonyStyle $io): void
+    {
+        /** @var ListableReceiverInterface $receiver */
+        $receiver = $this->getReceiver($failedTransportName);
+        $this->phpSerializer?->acceptPhpIncompleteClass();
+        try {
+            $envelope = $receiver->find($id);
+        } finally {
+            $this->phpSerializer?->rejectPhpIncompleteClass();
+        }
+        if (null === $envelope) {
+            throw new RuntimeException(\sprintf('The message "%s" was not found.', $id));
+        }
+
+        $this->displaySingleMessage($envelope, $io);
+
+        $io->writeln([
+            '',
+            \sprintf(' Run <comment>messenger:failed:retry %s --transport=%s</comment> to retry this message.', $id, $failedTransportName),
+            \sprintf(' Run <comment>messenger:failed:remove %s --transport=%s</comment> to delete it.', $id, $failedTransportName),
+        ]);
+    }
+}

vendor/symfony/messenger/Command/SetupTransportsCommand.php

@@ -0,0 +1,101 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Completion\CompletionInput;
+use Symfony\Component\Console\Completion\CompletionSuggestions;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Messenger\Transport\SetupableTransportInterface;
+
+/**
+ * @author Vincent Touzet <vincent.touzet@gmail.com>
+ */
+#[AsCommand(name: 'messenger:setup-transports', description: 'Prepare the required infrastructure for the transport')]
+class SetupTransportsCommand extends Command
+{
+    private ContainerInterface $transportLocator;
+    private array $transportNames;
+
+    public function __construct(ContainerInterface $transportLocator, array $transportNames = [])
+    {
+        $this->transportLocator = $transportLocator;
+        $this->transportNames = $transportNames;
+
+        parent::__construct();
+    }
+
+    /**
+     * @return void
+     */
+    protected function configure()
+    {
+        $this
+            ->addArgument('transport', InputArgument::OPTIONAL, 'Name of the transport to setup', null)
+            ->setHelp(<<<EOF
+The <info>%command.name%</info> command setups the transports:
+
+    <info>php %command.full_name%</info>
+
+Or a specific transport only:
+
+    <info>php %command.full_name% <transport></info>
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output);
+
+        $transportNames = $this->transportNames;
+        // do we want to set up only one transport?
+        if ($transport = $input->getArgument('transport')) {
+            if (!$this->transportLocator->has($transport)) {
+                throw new \RuntimeException(\sprintf('The "%s" transport does not exist.', $transport));
+            }
+            $transportNames = [$transport];
+        }
+
+        foreach ($transportNames as $id => $transportName) {
+            $transport = $this->transportLocator->get($transportName);
+            if (!$transport instanceof SetupableTransportInterface) {
+                $io->note(\sprintf('The "%s" transport does not support setup.', $transportName));
+                continue;
+            }
+
+            try {
+                $transport->setup();
+                $io->success(\sprintf('The "%s" transport was set up successfully.', $transportName));
+            } catch (\Exception $e) {
+                throw new \RuntimeException(\sprintf('An error occurred while setting up the "%s" transport: ', $transportName).$e->getMessage(), 0, $e);
+            }
+        }
+
+        return 0;
+    }
+
+    public function complete(CompletionInput $input, CompletionSuggestions $suggestions): void
+    {
+        if ($input->mustSuggestArgumentValuesFor('transport')) {
+            $suggestions->suggestValues($this->transportNames);
+
+            return;
+        }
+    }
+}

vendor/symfony/messenger/Command/StatsCommand.php

@@ -0,0 +1,95 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\ConsoleOutputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
+
+/**
+ * @author Kévin Thérage <therage.kevin@gmail.com>
+ */
+#[AsCommand(name: 'messenger:stats', description: 'Show the message count for one or more transports')]
+class StatsCommand extends Command
+{
+    private ContainerInterface $transportLocator;
+    private array $transportNames;
+
+    public function __construct(ContainerInterface $transportLocator, array $transportNames = [])
+    {
+        $this->transportLocator = $transportLocator;
+        $this->transportNames = $transportNames;
+
+        parent::__construct();
+    }
+
+    /**
+     * @return void
+     */
+    protected function configure()
+    {
+        $this
+            ->addArgument('transport_names', InputArgument::IS_ARRAY | InputArgument::OPTIONAL, 'List of transports\' names')
+            ->setHelp(<<<EOF
+The <info>%command.name%</info> command counts the messages for all the transports:
+
+    <info>php %command.full_name%</info>
+
+Or specific transports only:
+
+    <info>php %command.full_name% <transportNames></info>
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+
+        $transportNames = $this->transportNames;
+        if ($input->getArgument('transport_names')) {
+            $transportNames = $input->getArgument('transport_names');
+        }
+
+        $outputTable = [];
+        $uncountableTransports = [];
+        foreach ($transportNames as $transportName) {
+            if (!$this->transportLocator->has($transportName)) {
+                $io->warning(\sprintf('The "%s" transport does not exist.', $transportName));
+
+                continue;
+            }
+            $transport = $this->transportLocator->get($transportName);
+            if (!$transport instanceof MessageCountAwareInterface) {
+                $uncountableTransports[] = $transportName;
+
+                continue;
+            }
+            $outputTable[] = [$transportName, $transport->getMessageCount()];
+        }
+
+        $io->table(['Transport', 'Count'], $outputTable);
+
+        if ($uncountableTransports) {
+            $io->note(\sprintf('Unable to get message count for the following transports: "%s".', implode('", "', $uncountableTransports)));
+        }
+
+        return 0;
+    }
+}

vendor/symfony/messenger/Command/StopWorkersCommand.php

@@ -0,0 +1,67 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Command;
+
+use Psr\Cache\CacheItemPoolInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\ConsoleOutputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnRestartSignalListener;
+
+/**
+ * @author Ryan Weaver <ryan@symfonycasts.com>
+ */
+#[AsCommand(name: 'messenger:stop-workers', description: 'Stop workers after their current message')]
+class StopWorkersCommand extends Command
+{
+    private CacheItemPoolInterface $restartSignalCachePool;
+
+    public function __construct(CacheItemPoolInterface $restartSignalCachePool)
+    {
+        $this->restartSignalCachePool = $restartSignalCachePool;
+
+        parent::__construct();
+    }
+
+    protected function configure(): void
+    {
+        $this
+            ->setDefinition([])
+            ->setHelp(<<<'EOF'
+The <info>%command.name%</info> command sends a signal to stop any <info>messenger:consume</info> processes that are running.
+
+    <info>php %command.full_name%</info>
+
+Each worker command will finish the message they are currently processing
+and then exit. Worker commands are *not* automatically restarted: that
+should be handled by a process control system.
+EOF
+            )
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
+
+        $cacheItem = $this->restartSignalCachePool->getItem(StopWorkerOnRestartSignalListener::RESTART_REQUESTED_TIMESTAMP_KEY);
+        $cacheItem->set(microtime(true));
+        $this->restartSignalCachePool->save($cacheItem);
+
+        $io->success('Signal successfully sent to stop any running workers.');
+
+        return 0;
+    }
+}