src/Symfony/Component/Messenger/Command/AbstractFailedMessagesCommand.php - 8.2 - githubmirror/symfony

src/Symfony/Component/Messenger/Command/AbstractFailedMessagesCommand.php
295 строк12 KB
Nicolas Grekas
[Messenger] Filter failed messages by class and failure time
01 авг 2026, 10:06
01 авг 2026, 10:06b19cf8f
<?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\Exception\RuntimeException;
use Symfony\Component\Console\Helper\Dumper;
use Symfony\Component\Console\Input\InputInterface;
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\Exception\MessageDecodingFailedException;
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';

    public function __construct(
        private ?string $globalFailureReceiverName,
        /**
         * @var ServiceProviderInterface<ReceiverInterface>
         */
        protected ServiceProviderInterface $failureTransports,
        protected ?PhpSerializer $phpSerializer = null,
    ) {
        parent::__construct();
    }

    protected function getGlobalFailureReceiverName(): ?string
    {
        return $this->globalFailureReceiverName;
    }

    protected function getMessageId(Envelope $envelope): mixed
    {
        $stamp = $envelope->last(TransportMessageIdStamp::class);

        return $stamp?->getId();
    }

    protected function displaySingleMessage(Envelope $envelope, SymfonyStyle $io, ?SymfonyStyle $errorIo = null): void
    {
        $errorIo ??= $io->getErrorStyle();

        $io->title('Failed Message Details');

        $messageClass = $envelope->getMessage()::class;
        $lastErrorDetailsStamp = $envelope->last(ErrorDetailsStamp::class);
        $lastMessageDecodingFailed = MessageDecodingFailedException::class === $messageClass || $envelope->last(MessageDecodingFailedStamp::class);

        $rows = [
            ['Class', $messageClass],
        ];

        if (null !== $id = $this->getMessageId($envelope)) {
            $rows[] = ['Message Id', $id];
        }

        if (!$sentToFailureTransportStamp = $envelope->last(SentToFailureTransportStamp::class)) {
            $errorIo->warning('Message does not appear to have been sent to this transport after failing');
        } else {
            $rows = array_merge($rows, [
                ['Failed at', $envelope->last(RedeliveryStamp::class)?->getRedeliveredAt()->format('Y-m-d H:i:s') ?? ''],
                ['Error', $lastErrorDetailsStamp?->getExceptionMessage() ?? ''],
                ['Error Code', $lastErrorDetailsStamp?->getExceptionCode() ?? ''],
                ['Error Class', $lastErrorDetailsStamp?->getExceptionClass() ?? '(unknown)'],
                ['Transport', $sentToFailureTransportStamp->getOriginalReceiverName()],
            ]);
        }

        $io->table([], $rows);

        $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 ($lastMessageDecodingFailed) {
                $errorIo->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 ($lastMessageDecodingFailed) {
                $errorIo->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 <info>1</info> message pending in the failure transport.');
            } else {
                $io->writeln(\sprintf('There are <info>%d</info> messages pending in the failure transport.', $receiver->getMessageCount()));
            }
        }
    }

    /**
     * @param bool $hasIds Whether explicit message ids were given, which the filters cannot be combined with
     *
     * @return array{?string, ?\DateTimeImmutable, ?\DateTimeImmutable} The class name, the earliest and the latest failure time to select
     */
    protected function getFilters(InputInterface $input, bool $hasIds): array
    {
        $classFilter = $input->getOption('class-filter');
        $failedAfter = $this->getDateOption($input, 'failed-after');
        $failedBefore = $this->getDateOption($input, 'failed-before');

        if ($hasIds && (null !== $classFilter || null !== $failedAfter || null !== $failedBefore)) {
            throw new RuntimeException('You cannot specify message ids when using the "--class-filter", "--failed-after" or "--failed-before" options.');
        }

        return [$classFilter, $failedAfter, $failedBefore];
    }

    /**
     * @return list<mixed> The ids of the messages matching every given filter
     */
    protected function getMessageIdsByFilter(ListableReceiverInterface $receiver, ?string $classFilter, ?\DateTimeImmutable $failedAfter, ?\DateTimeImmutable $failedBefore): array
    {
        $ids = [];

        $this->phpSerializer?->acceptPhpIncompleteClass();
        try {
            foreach ($receiver->all() as $envelope) {
                if ($this->matchesFilter($envelope, $classFilter, $failedAfter, $failedBefore)) {
                    $ids[] = $this->getMessageId($envelope);
                }
            }
        } finally {
            $this->phpSerializer?->rejectPhpIncompleteClass();
        }

        return $ids;
    }

    protected function matchesFilter(Envelope $envelope, ?string $classFilter, ?\DateTimeImmutable $failedAfter, ?\DateTimeImmutable $failedBefore): bool
    {
        if (null !== $classFilter && $classFilter !== $envelope->getMessage()::class) {
            return false;
        }

        if (null === $failedAfter && null === $failedBefore) {
            return true;
        }

        // messages that were never redelivered have no known failure time, so no time window can select them
        if (null === $failedAt = $envelope->last(RedeliveryStamp::class)?->getRedeliveredAt()) {
            return false;
        }

        return (null === $failedAfter || $failedAt >= $failedAfter) && (null === $failedBefore || $failedAt <= $failedBefore);
    }

    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 getDateOption(InputInterface $input, string $option): ?\DateTimeImmutable
    {
        if (null === $value = $input->getOption($option)) {
            return null;
        }

        try {
            return new \DateTimeImmutable($value);
        } catch (\DateMalformedStringException $e) {
            throw new InvalidArgumentException(\sprintf('The value of the "--%s" option is not a valid date: "%s".', $option, $value), previous: $e);
        }
    }

    private function createCloner(): ?ClonerInterface
    {
        if (!class_exists(VarCloner::class)) {
            return null;
        }

        $cloner = new VarCloner();
        $cloner->addCasters([FlattenException::class => static 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()),
                Caster::PREFIX_VIRTUAL.'previous' => $flattenException->getPrevious(),
            ];
        }]);

        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 <info>global</info> failure transport <info>%s</info>.', $failureTransportName),
                '> To use a different failure transport, pass <info>--transport=</info>.',
                \sprintf('> Available failure transports are: <info>%s</info>', 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);
        }
    }
}