vendor/symfony/messenger/Command/FailedMessagesRetryCommand.php line 45

  1. <?php
  3. /*
  4.  * This file is part of the Symfony package.
  5.  *
  6.  * (c) Fabien Potencier <fabien@symfony.com>
  7.  *
  8.  * For the full copyright and license information, please view the LICENSE
  9.  * file that was distributed with this source code.
  10.  */
  12. namespace Symfony\Component\Messenger\Command;
  14. use Psr\Log\LoggerInterface;
  15. use Symfony\Component\Console\Attribute\AsCommand;
  16. use Symfony\Component\Console\Exception\RuntimeException;
  17. use Symfony\Component\Console\Input\InputArgument;
  18. use Symfony\Component\Console\Input\InputInterface;
  19. use Symfony\Component\Console\Input\InputOption;
  20. use Symfony\Component\Console\Output\ConsoleOutputInterface;
  21. use Symfony\Component\Console\Output\OutputInterface;
  22. use Symfony\Component\Console\Style\SymfonyStyle;
  23. use Symfony\Component\EventDispatcher\EventDispatcherInterface;
  24. use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
  25. use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
  26. use Symfony\Component\Messenger\MessageBusInterface;
  27. use Symfony\Component\Messenger\Stamp\MessageDecodingFailedStamp;
  28. use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
  29. use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
  30. use Symfony\Component\Messenger\Transport\Receiver\SingleMessageReceiver;
  31. use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
  32. use Symfony\Component\Messenger\Worker;
  33. use Symfony\Contracts\Service\ServiceProviderInterface;
  35. /**
  36.  * @author Ryan Weaver <ryan@symfonycasts.com>
  37.  */
  38. #[AsCommand(name: 'messenger:failed:retry', description: 'Retry one or more messages from the failure transport')]
  39. class FailedMessagesRetryCommand extends AbstractFailedMessagesCommand
  40. {
  41.     private EventDispatcherInterface $eventDispatcher;
  42.     private MessageBusInterface $messageBus;
  43.     private ?LoggerInterface $logger;
  45.     public function __construct(?string $globalReceiverName, ServiceProviderInterface $failureTransports, MessageBusInterface $messageBus, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null, PhpSerializer $phpSerializer = null)
  46.     {
  47.         $this->eventDispatcher = $eventDispatcher;
  48.         $this->messageBus = $messageBus;
  49.         $this->logger = $logger;
  51.         parent::__construct($globalReceiverName, $failureTransports, $phpSerializer);
  52.     }
  54.     protected function configure(): void
  55.     {
  56.         $this
  57.             ->setDefinition([
  58.                 new InputArgument('id', InputArgument::IS_ARRAY, 'Specific message id(s) to retry'),
  59.                 new InputOption('force', null, InputOption::VALUE_NONE, 'Force action without confirmation'),
  60.                 new InputOption('transport', null, InputOption::VALUE_OPTIONAL, 'Use a specific failure transport', self::DEFAULT_TRANSPORT_OPTION),
  61.             ])
  62.             ->setHelp(<<<'EOF'
  63. The <info>%command.name%</info> retries message in the failure transport.
  65.     <info>php %command.full_name%</info>
  67. The command will interactively ask if each message should be retried
  68. or discarded.
  70. Some transports support retrying a specific message id, which comes
  71. from the <info>messenger:failed:show</info> command.
  73.     <info>php %command.full_name% {id}</info>
  75. Or pass multiple ids at once to process multiple messages:
  77. <info>php %command.full_name% {id1} {id2} {id3}</info>
  79. EOF
  80.             )
  81.         ;
  82.     }
  84.     protected function execute(InputInterface $input, OutputInterface $output): int
  85.     {
  86.         $this->eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1));
  88.         $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
  89.         $io->comment('Quit this command with CONTROL-C.');
  90.         if (!$output->isVeryVerbose()) {
  91.             $io->comment('Re-run the command with a -vv option to see logs about consumed messages.');
  92.         }
  94.         $failureTransportName = $input->getOption('transport');
  95.         if (self::DEFAULT_TRANSPORT_OPTION === $failureTransportName) {
  96.             $this->printWarningAvailableFailureTransports($io, $this->getGlobalFailureReceiverName());
  97.         }
  98.         if ('' === $failureTransportName || null === $failureTransportName) {
  99.             $failureTransportName = $this->interactiveChooseFailureTransport($io);
  100.         }
  101.         $failureTransportName = self::DEFAULT_TRANSPORT_OPTION === $failureTransportName ? $this->getGlobalFailureReceiverName() : $failureTransportName;
  103.         $receiver = $this->getReceiver($failureTransportName);
  104.         $this->printPendingMessagesMessage($receiver, $io);
  106.         $io->writeln(sprintf('To retry all the messages, run <comment>messenger:consume %s</comment>', $failureTransportName));
  108.         $shouldForce = $input->getOption('force');
  109.         $ids = $input->getArgument('id');
  110.         if (0 === \count($ids)) {
  111.             if (!$input->isInteractive()) {
  112.                 throw new RuntimeException('Message id must be passed when in non-interactive mode.');
  113.             }
  115.             $this->runInteractive($failureTransportName, $io, $shouldForce);
  117.             return 0;
  118.         }
  120.         $this->retrySpecificIds($failureTransportName, $ids, $io, $shouldForce);
  121.         $io->success('All done!');
  123.         return 0;
  124.     }
  126.     private function runInteractive(string $failureTransportName, SymfonyStyle $io, bool $shouldForce)
  127.     {
  128.         $receiver = $this->failureTransports->get($failureTransportName);
  129.         $count = 0;
  130.         if ($receiver instanceof ListableReceiverInterface) {
  131.             // for listable receivers, find the messages one-by-one
  132.             // this avoids using get(), which for some less-robust
  133.             // transports (like Doctrine), will cause the message
  134.             // to be temporarily "acked", even if the user aborts
  135.             // handling the message
  136.             while (true) {
  137.                 $envelopes = [];
  138.                 $this->phpSerializer?->acceptPhpIncompleteClass();
  139.                 try {
  140.                     foreach ($receiver->all(1) as $envelope) {
  141.                         ++$count;
  142.                         $envelopes[] = $envelope;
  143.                     }
  144.                 } finally {
  145.                     $this->phpSerializer?->rejectPhpIncompleteClass();
  146.                 }
  148.                 // break the loop if all messages are consumed
  149.                 if (0 === \count($envelopes)) {
  150.                     break;
  151.                 }
  153.                 $this->retrySpecificEnvelopes($envelopes, $failureTransportName, $io, $shouldForce);
  154.             }
  155.         } else {
  156.             // get() and ask messages one-by-one
  157.             $count = $this->runWorker($failureTransportName, $receiver, $io, $shouldForce);
  158.         }
  160.         // avoid success message if nothing was processed
  161.         if (1 <= $count) {
  162.             $io->success('All failed messages have been handled or removed!');
  163.         }
  164.     }
  166.     private function runWorker(string $failureTransportName, ReceiverInterface $receiver, SymfonyStyle $io, bool $shouldForce): int
  167.     {
  168.         $count = 0;
  169.         $listener = function (WorkerMessageReceivedEvent $messageReceivedEvent) use ($io, $receiver, $shouldForce, &$count) {
  170.             ++$count;
  171.             $envelope = $messageReceivedEvent->getEnvelope();
  173.             $this->displaySingleMessage($envelope, $io);
  175.             if ($envelope->last(MessageDecodingFailedStamp::class)) {
  176.                 throw new \RuntimeException(sprintf('The message with id "%s" could not decoded, it can only be shown or removed.', $this->getMessageId($envelope) ?? '?'));
  177.             }
  179.             $shouldHandle = $shouldForce || $io->confirm('Do you want to retry (yes) or delete this message (no)?');
  181.             if ($shouldHandle) {
  182.                 return;
  183.             }
  185.             $messageReceivedEvent->shouldHandle(false);
  186.             $receiver->reject($envelope);
  187.         };
  188.         $this->eventDispatcher->addListener(WorkerMessageReceivedEvent::class, $listener);
  190.         $worker = new Worker(
  191.             [$failureTransportName => $receiver],
  192.             $this->messageBus,
  193.             $this->eventDispatcher,
  194.             $this->logger
  195.         );
  197.         try {
  198.             $worker->run();
  199.         } finally {
  200.             $this->eventDispatcher->removeListener(WorkerMessageReceivedEvent::class, $listener);
  201.         }
  203.         return $count;
  204.     }
  206.     private function retrySpecificIds(string $failureTransportName, array $ids, SymfonyStyle $io, bool $shouldForce)
  207.     {
  208.         $receiver = $this->getReceiver($failureTransportName);
  210.         if (!$receiver instanceof ListableReceiverInterface) {
  211.             throw new RuntimeException(sprintf('The "%s" receiver does not support retrying messages by id.', $failureTransportName));
  212.         }
  214.         foreach ($ids as $id) {
  215.             $this->phpSerializer?->acceptPhpIncompleteClass();
  216.             try {
  217.                 $envelope = $receiver->find($id);
  218.             } finally {
  219.                 $this->phpSerializer?->rejectPhpIncompleteClass();
  220.             }
  221.             if (null === $envelope) {
  222.                 throw new RuntimeException(sprintf('The message "%s" was not found.', $id));
  223.             }
  225.             $singleReceiver = new SingleMessageReceiver($receiver, $envelope);
  226.             $this->runWorker($failureTransportName, $singleReceiver, $io, $shouldForce);
  227.         }
  228.     }
  230.     private function retrySpecificEnvelopes(array $envelopes, string $failureTransportName, SymfonyStyle $io, bool $shouldForce)
  231.     {
  232.         $receiver = $this->getReceiver($failureTransportName);
  234.         foreach ($envelopes as $envelope) {
  235.             $singleReceiver = new SingleMessageReceiver($receiver, $envelope);
  236.             $this->runWorker($failureTransportName, $singleReceiver, $io, $shouldForce);
  237.         }
  238.     }
  239. }