0

Final Class Yiisoft\Queue\Command\ListenAllCommand

All Classes | Methods
InheritanceYiisoft\Queue\Command\ListenAllCommand » Symfony\Component\Console\Command\Command

Public Methods

Hide inherited methods

Method Description Defined By
__construct() Yiisoft\Queue\Command\ListenAllCommand
configure() Yiisoft\Queue\Command\ListenAllCommand

Protected Methods

Hide inherited methods

Method Description Defined By
execute() Yiisoft\Queue\Command\ListenAllCommand

Method Details

Hide inherited methods

__construct() public method

public mixed __construct ( Yiisoft\Queue\Provider\QueueConsumerProviderInterface $queueProvider, Yiisoft\Queue\Cli\LoopInterface $loop )
$queueProvider Yiisoft\Queue\Provider\QueueConsumerProviderInterface
$loop Yiisoft\Queue\Cli\LoopInterface 

                public function __construct(
    private readonly QueueConsumerProviderInterface $queueProvider,
    private readonly LoopInterface $loop,
) {
    parent::__construct();
}
configure() public method

public void configure ( )

                public function configure(): void
{
    $this->addArgument(
        'queue',
        InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
        'Queue name list to connect to',
        [],
    )
        ->addOption(
            'pause',
            'p',
            InputOption::VALUE_REQUIRED,
            'Pause between queue iterations in seconds. May save some CPU. Default: 1',
            1,
        )
        ->addOption(
            'limit',
            'm',
            InputOption::VALUE_REQUIRED,
            'Maximum number of messages to process in each queue before switching to another queue. '
                . 'Default is 0 (no limits).',
            0,
        );
    $this->addUsage('[queue1 [queue2 [...]]] [--pause=<pause>] [--limit=<limit>]');
}
execute() protected method

protected integer execute ( \Symfony\Component\Console\Input\InputInterface $input, \Symfony\Component\Console\Output\OutputInterface $output )
$input \Symfony\Component\Console\Input\InputInterface
$output \Symfony\Component\Console\Output\OutputInterface 

                protected function execute(InputInterface $input, OutputInterface $output): int
{
    /** @var string[] $queueNames */
    $queueNames = $input->getArgument('queue');
    if ($queueNames === []) {
        $queueNames = $this->queueProvider->getConsumerQueueNames();
    }
    $consumers = [];
    /** @var string $queueName */
    foreach ($queueNames as $queueName) {
        $consumers[] = $this->queueProvider->getConsumer($queueName);
    }
    if ($consumers === []) {
        $output->writeln('No consumers are configured.');
        return Command::SUCCESS;
    }
    $pauseSeconds = (int) $input->getOption('pause');
    if ($pauseSeconds < 0) {
        $pauseSeconds = 1;
    }
    while ($this->loop->canContinue()) {
        $hasMessages = false;
        foreach ($consumers as $consumer) {
            $hasMessages = $consumer->run((int) $input->getOption('limit')) > 0 || $hasMessages;
        }
        if (!$hasMessages) {
            /** @psalm-var 0|positive-int $pauseSeconds */
            sleep($pauseSeconds);
        }
    }
    return 0;
}