vendor/symfony/messenger/Command/ConsumeMessagesCommand.php line 53

  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\Container\ContainerInterface;
  15. use Psr\Log\LoggerInterface;
  16. use Symfony\Component\Console\Attribute\AsCommand;
  17. use Symfony\Component\Console\Command\Command;
  18. use Symfony\Component\Console\Completion\CompletionInput;
  19. use Symfony\Component\Console\Completion\CompletionSuggestions;
  20. use Symfony\Component\Console\Exception\InvalidOptionException;
  21. use Symfony\Component\Console\Exception\RuntimeException;
  22. use Symfony\Component\Console\Input\InputArgument;
  23. use Symfony\Component\Console\Input\InputInterface;
  24. use Symfony\Component\Console\Input\InputOption;
  25. use Symfony\Component\Console\Output\ConsoleOutputInterface;
  26. use Symfony\Component\Console\Output\OutputInterface;
  27. use Symfony\Component\Console\Question\ChoiceQuestion;
  28. use Symfony\Component\Console\Style\SymfonyStyle;
  29. use Symfony\Component\EventDispatcher\EventDispatcherInterface;
  30. use Symfony\Component\Messenger\EventListener\ResetServicesListener;
  31. use Symfony\Component\Messenger\EventListener\StopWorkerOnFailureLimitListener;
  32. use Symfony\Component\Messenger\EventListener\StopWorkerOnMemoryLimitListener;
  33. use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
  34. use Symfony\Component\Messenger\EventListener\StopWorkerOnTimeLimitListener;
  35. use Symfony\Component\Messenger\RoutableMessageBus;
  36. use Symfony\Component\Messenger\Worker;
  38. /**
  39.  * @author Samuel Roze <samuel.roze@gmail.com>
  40.  */
  41. #[AsCommand(name: 'messenger:consume', description: 'Consume messages')]
  42. class ConsumeMessagesCommand extends Command
  43. {
  44.     private RoutableMessageBus $routableBus;
  45.     private ContainerInterface $receiverLocator;
  46.     private EventDispatcherInterface $eventDispatcher;
  47.     private ?LoggerInterface $logger;
  48.     private array $receiverNames;
  49.     private ?ResetServicesListener $resetServicesListener;
  50.     private array $busIds;
  51.     private ?ContainerInterface $rateLimiterLocator;
  53.     public function __construct(RoutableMessageBus $routableBus, ContainerInterface $receiverLocator, EventDispatcherInterface $eventDispatcher, LoggerInterface $logger = null, array $receiverNames = [], ResetServicesListener $resetServicesListener = null, array $busIds = [], ContainerInterface $rateLimiterLocator = null)
  54.     {
  55.         $this->routableBus = $routableBus;
  56.         $this->receiverLocator = $receiverLocator;
  57.         $this->eventDispatcher = $eventDispatcher;
  58.         $this->logger = $logger;
  59.         $this->receiverNames = $receiverNames;
  60.         $this->resetServicesListener = $resetServicesListener;
  61.         $this->busIds = $busIds;
  62.         $this->rateLimiterLocator = $rateLimiterLocator;
  64.         parent::__construct();
  65.     }
  67.     protected function configure(): void
  68.     {
  69.         $defaultReceiverName = 1 === \count($this->receiverNames) ? current($this->receiverNames) : null;
  71.         $this
  72.             ->setDefinition([
  73.                 new InputArgument('receivers', InputArgument::IS_ARRAY, 'Names of the receivers/transports to consume in order of priority', $defaultReceiverName ? [$defaultReceiverName] : []),
  74.                 new InputOption('limit', 'l', InputOption::VALUE_REQUIRED, 'Limit the number of received messages'),
  75.                 new InputOption('failure-limit', 'f', InputOption::VALUE_REQUIRED, 'The number of failed messages the worker can consume'),
  76.                 new InputOption('memory-limit', 'm', InputOption::VALUE_REQUIRED, 'The memory limit the worker can consume'),
  77.                 new InputOption('time-limit', 't', InputOption::VALUE_REQUIRED, 'The time limit in seconds the worker can handle new messages'),
  78.                 new InputOption('sleep', null, InputOption::VALUE_REQUIRED, 'Seconds to sleep before asking for new messages after no messages were found', 1),
  79.                 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)'),
  80.                 new InputOption('queues', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Limit receivers to only consume from the specified queues'),
  81.                 new InputOption('no-reset', null, InputOption::VALUE_NONE, 'Do not reset container services after each message'),
  82.             ])
  83.             ->setHelp(<<<'EOF'
  84. The <info>%command.name%</info> command consumes messages and dispatches them to the message bus.
  86.     <info>php %command.full_name% <receiver-name></info>
  88. To receive from multiple transports, pass each name:
  90.     <info>php %command.full_name% receiver1 receiver2</info>
  92. Use the --limit option to limit the number of messages received:
  94.     <info>php %command.full_name% <receiver-name> --limit=10</info>
  96. Use the --failure-limit option to stop the worker when the given number of failed messages is reached:
  98.     <info>php %command.full_name% <receiver-name> --failure-limit=2</info>
  100. 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]:
  102.     <info>php %command.full_name% <receiver-name> --memory-limit=128M</info>
  104. Use the --time-limit option to stop the worker when the given time limit (in seconds) is reached.
  105. If a message is being handled, the worker will stop after the processing is finished:
  107.     <info>php %command.full_name% <receiver-name> --time-limit=3600</info>
  109. Use the --bus option to specify the message bus to dispatch received messages
  110. to instead of trying to determine it automatically. This is required if the
  111. messages didn't originate from Messenger:
  113.     <info>php %command.full_name% <receiver-name> --bus=event_bus</info>
  115. Use the --queues option to limit a receiver to only certain queues (only supported by some receivers):
  117.     <info>php %command.full_name% <receiver-name> --queues=fasttrack</info>
  119. Use the --no-reset option to prevent services resetting after each message (may lead to leaking services' state between messages):
  121.     <info>php %command.full_name% <receiver-name> --no-reset</info>
  122. EOF
  123.             )
  124.         ;
  125.     }
  127.     protected function interact(InputInterface $input, OutputInterface $output)
  128.     {
  129.         $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
  131.         if ($this->receiverNames && !$input->getArgument('receivers')) {
  132.             $io->block('Which transports/receivers do you want to consume?', null, 'fg=white;bg=blue', ' ', true);
  134.             $io->writeln('Choose which receivers you want to consume messages from in order of priority.');
  135.             if (\count($this->receiverNames) > 1) {
  136.                 $io->writeln(sprintf('Hint: to consume from multiple, use a list of their names, e.g. <comment>%s</comment>', implode(', ', $this->receiverNames)));
  137.             }
  139.             $question = new ChoiceQuestion('Select receivers to consume:', $this->receiverNames, 0);
  140.             $question->setMultiselect(true);
  142.             $input->setArgument('receivers', $io->askQuestion($question));
  143.         }
  145.         if (!$input->getArgument('receivers')) {
  146.             throw new RuntimeException('Please pass at least one receiver.');
  147.         }
  148.     }
  150.     protected function execute(InputInterface $input, OutputInterface $output): int
  151.     {
  152.         $receivers = [];
  153.         $rateLimiters = [];
  154.         foreach ($receiverNames = $input->getArgument('receivers') as $receiverName) {
  155.             if (!$this->receiverLocator->has($receiverName)) {
  156.                 $message = sprintf('The receiver "%s" does not exist.', $receiverName);
  157.                 if ($this->receiverNames) {
  158.                     $message .= sprintf(' Valid receivers are: %s.', implode(', ', $this->receiverNames));
  159.                 }
  161.                 throw new RuntimeException($message);
  162.             }
  164.             $receivers[$receiverName] = $this->receiverLocator->get($receiverName);
  165.             if ($this->rateLimiterLocator?->has($receiverName)) {
  166.                 $rateLimiters[$receiverName] = $this->rateLimiterLocator->get($receiverName);
  167.             }
  168.         }
  170.         if (null !== $this->resetServicesListener && !$input->getOption('no-reset')) {
  171.             $this->eventDispatcher->addSubscriber($this->resetServicesListener);
  172.         }
  174.         $stopsWhen = [];
  175.         if (null !== $limit = $input->getOption('limit')) {
  176.             if (!is_numeric($limit) || 0 >= $limit) {
  177.                 throw new InvalidOptionException(sprintf('Option "limit" must be a positive integer, "%s" passed.', $limit));
  178.             }
  180.             $stopsWhen[] = "processed {$limit} messages";
  181.             $this->eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener($limit, $this->logger));
  182.         }
  184.         if ($failureLimit = $input->getOption('failure-limit')) {
  185.             $stopsWhen[] = "reached {$failureLimit} failed messages";
  186.             $this->eventDispatcher->addSubscriber(new StopWorkerOnFailureLimitListener($failureLimit, $this->logger));
  187.         }
  189.         if ($memoryLimit = $input->getOption('memory-limit')) {
  190.             $stopsWhen[] = "exceeded {$memoryLimit} of memory";
  191.             $this->eventDispatcher->addSubscriber(new StopWorkerOnMemoryLimitListener($this->convertToBytes($memoryLimit), $this->logger));
  192.         }
  194.         if (null !== $timeLimit = $input->getOption('time-limit')) {
  195.             if (!is_numeric($timeLimit) || 0 >= $timeLimit) {
  196.                 throw new InvalidOptionException(sprintf('Option "time-limit" must be a positive integer, "%s" passed.', $timeLimit));
  197.             }
  199.             $stopsWhen[] = "been running for {$timeLimit}s";
  200.             $this->eventDispatcher->addSubscriber(new StopWorkerOnTimeLimitListener($timeLimit, $this->logger));
  201.         }
  203.         $stopsWhen[] = 'received a stop signal via the messenger:stop-workers command';
  205.         $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
  206.         $io->success(sprintf('Consuming messages from transport%s "%s".', \count($receivers) > 1 ? 's' : '', implode(', ', $receiverNames)));
  208.         if ($stopsWhen) {
  209.             $last = array_pop($stopsWhen);
  210.             $stopsWhen = ($stopsWhen ? implode(', ', $stopsWhen).' or ' : '').$last;
  211.             $io->comment("The worker will automatically exit once it has {$stopsWhen}.");
  212.         }
  214.         $io->comment('Quit the worker with CONTROL-C.');
  216.         if (OutputInterface::VERBOSITY_VERBOSE > $output->getVerbosity()) {
  217.             $io->comment('Re-run the command with a -vv option to see logs about consumed messages.');
  218.         }
  220.         $bus = $input->getOption('bus') ? $this->routableBus->getMessageBus($input->getOption('bus')) : $this->routableBus;
  222.         $worker = new Worker($receivers, $bus, $this->eventDispatcher, $this->logger, $rateLimiters);
  223.         $options = [
  224.             'sleep' => $input->getOption('sleep') * 1000000,
  225.         ];
  226.         if ($queues = $input->getOption('queues')) {
  227.             $options['queues'] = $queues;
  228.         }
  229.         $worker->run($options);
  231.         return 0;
  232.     }
  234.     public function complete(CompletionInput $input, CompletionSuggestions $suggestions): void
  235.     {
  236.         if ($input->mustSuggestArgumentValuesFor('receivers')) {
  237.             $suggestions->suggestValues(array_diff($this->receiverNames, array_diff($input->getArgument('receivers'), [$input->getCompletionValue()])));
  239.             return;
  240.         }
  242.         if ($input->mustSuggestOptionValuesFor('bus')) {
  243.             $suggestions->suggestValues($this->busIds);
  244.         }
  245.     }
  247.     private function convertToBytes(string $memoryLimit): int
  248.     {
  249.         $memoryLimit = strtolower($memoryLimit);
  250.         $max = ltrim($memoryLimit, '+');
  251.         if (str_starts_with($max, '0x')) {
  252.             $max = \intval($max, 16);
  253.         } elseif (str_starts_with($max, '0')) {
  254.             $max = \intval($max, 8);
  255.         } else {
  256.             $max = (int) $max;
  257.         }
  259.         switch (substr(rtrim($memoryLimit, 'b'), -1)) {
  260.             case 't': $max *= 1024;
  261.                 // no break
  262.             case 'g': $max *= 1024;
  263.                 // no break
  264.             case 'm': $max *= 1024;
  265.                 // no break
  266.             case 'k': $max *= 1024;
  267.         }
  269.         return $max;
  270.     }
  271. }