WorkerCommand.php

<?php
declare(strict_types=1);

/**
 * CakePHP(tm) : Rapid Development Framework (https://cakephp.org)
 * Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
 *
 * Licensed under The MIT License
 * For full copyright and license information, please see the LICENSE.txt
 * Redistributions of files must retain the above copyright notice.
 *
 * @copyright     Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
 * @link          https://cakephp.org CakePHP(tm) Project
 * @since         0.1.0
 * @license       https://opensource.org/licenses/MIT MIT License
 */
namespace Cake\Queue\Command;

use Cake\Command\Command;
use Cake\Console\Arguments;
use Cake\Console\ConsoleIo;
use Cake\Console\ConsoleOptionParser;
use Cake\Core\Configure;
use Cake\Core\ContainerInterface;
use Cake\Log\Log;
use Cake\Queue\Consumption\LimitAttemptsExtension;
use Cake\Queue\Consumption\LimitConsumedMessagesExtension;
use Cake\Queue\Consumption\RemoveUniqueJobIdFromCacheExtension;
use Cake\Queue\Listener\FailedJobsListener;
use Cake\Queue\Queue\Processor;
use Cake\Queue\Queue\SubprocessProcessor;
use Cake\Queue\QueueManager;
use DateTime;
use Enqueue\Consumption\ChainExtension;
use Enqueue\Consumption\Extension\LimitConsumptionTimeExtension;
use Enqueue\Consumption\Extension\LoggerExtension;
use Enqueue\Consumption\ExtensionInterface;
use Interop\Queue\Processor as InteropProcessor;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

/**
 * Worker command.
 */
class WorkerCommand extends Command
{
    /**
     * @param \Cake\Core\ContainerInterface|null $container DI container instance
     */
    public function __construct(
        protected readonly ?ContainerInterface $container = null,
    ) {
    }

    /**
     * Get the command name.
     *
     * @return string
     */
    public static function defaultName(): string
    {
        return 'queue worker';
    }

    /**
     * Gets the option parser instance and configures it.
     *
     * @return \Cake\Console\ConsoleOptionParser
     */
    public function getOptionParser(): ConsoleOptionParser
    {
        $parser = parent::getOptionParser();

        $parser->addOption('config', [
            'default' => 'default',
            'help' => 'Name of a queue config to use',
            'short' => 'c',
        ]);
        $parser->addOption('queue', [
            'help' => 'Name of queue to bind to. Defaults to the queue config (--config).',
            'short' => 'Q',
        ]);
        $parser->addOption('processor', [
            'help' => 'Name of processor to bind to',
            'default' => null,
            'short' => 'p',
        ]);
        $parser->addOption('logger', [
            'help' => 'Name of a configured logger',
            'default' => 'stdout',
            'short' => 'l',
        ]);
        $parser->addOption('max-jobs', [
            'help' => 'Maximum number of jobs to process. Worker will exit after limit is reached.',
            'default' => null,
            'short' => 'i',
        ]);
        $parser->addOption('max-runtime', [
            'help' => 'Maximum number of seconds worker will run. Worker will exit after limit is reached.',
            'default' => null,
            'short' => 'r',
        ]);
        $parser->addOption('max-attempts', [
            'help' => 'Maximum number of times each job will be attempted.'
                . ' Maximum attempts defined on a job will override this value.',
            'default' => null,
            'short' => 'a',
        ]);
        $parser->addOption('subprocess', [
            'help' => 'Execute jobs in a subprocess. Useful for development to reload code for each job.',
            'boolean' => true,
            'default' => false,
        ]);
        $parser->setDescription(
            'Runs a queue worker that consumes from the named queue.',
        );

        return $parser;
    }

    /**
     * Creates and returns a QueueExtension object
     *
     * @param \Cake\Console\Arguments $args Arguments
     * @param \Psr\Log\LoggerInterface $logger Logger instance.
     * @return \Enqueue\Consumption\ExtensionInterface
     */
    protected function getQueueExtension(Arguments $args, LoggerInterface $logger): ExtensionInterface
    {
        $limitAttempsExtension = new LimitAttemptsExtension((int)$args->getOption('max-attempts') ?: null);

        $limitAttempsExtension->getEventManager()->on(new FailedJobsListener());

        $configKey = (string)$args->getOption('config');
        $config = QueueManager::getConfig($configKey);

        $extensions = [
            new LoggerExtension($logger),
            $limitAttempsExtension,
        ];

        if ($args->getOption('max-jobs') !== null) {
            $maxJobs = (int)$args->getOption('max-jobs');
            $extensions[] = new LimitConsumedMessagesExtension($maxJobs);
        }

        if ($args->getOption('max-runtime') !== null) {
            $endTime = new DateTime(sprintf('+%d seconds', (int)$args->getOption('max-runtime')));
            $extensions[] = new LimitConsumptionTimeExtension($endTime);
        }

        if (isset($config['uniqueCacheKey'])) {
            $extensions[] = new RemoveUniqueJobIdFromCacheExtension($config['uniqueCacheKey']);
        }

        return new ChainExtension($extensions);
    }

    /**
     * Creates and returns a LoggerInterface object
     *
     * @param \Cake\Console\Arguments $args Arguments
     * @param bool $forceLogger Force logger creation even without verbose flag
     * @return \Psr\Log\LoggerInterface
     */
    protected function getLogger(Arguments $args, bool $forceLogger = false): LoggerInterface
    {
        $logger = null;
        if ($forceLogger || !empty($args->getOption('verbose'))) {
            $logger = Log::engine((string)$args->getOption('logger'));
        }

        return $logger ?? new NullLogger();
    }

    /**
     * Creates and returns a Processor object
     *
     * @param \Cake\Console\Arguments $args Arguments
     * @param \Cake\Console\ConsoleIo $io ConsoleIo
     * @param \Psr\Log\LoggerInterface $logger Logger instance
     * @return \Interop\Queue\Processor
     */
    protected function getProcessor(Arguments $args, ConsoleIo $io, LoggerInterface $logger): InteropProcessor
    {
        $configKey = (string)$args->getOption('config');
        $config = QueueManager::getConfig($configKey);

        $processorClass = $config['processor'] ?? Processor::class;

        if (!class_exists($processorClass)) {
            $io->error(sprintf('Processor class %s not found', $processorClass));
            $this->abort();
        }

        if (!is_subclass_of($processorClass, InteropProcessor::class)) {
            $io->error(sprintf('Processor class %s must implement Interop\Queue\Processor', $processorClass));
            $this->abort();
        }

        // Check subprocess mode before instantiating processor
        if ($args->getOption('subprocess') || ($config['subprocess']['enabled'] ?? false)) {
            if ($processorClass !== Processor::class && !is_subclass_of($processorClass, Processor::class)) {
                $io->error('Subprocess mode is only supported with the default Processor class');
                $this->abort();
            }

            $subprocessConfig = array_merge(
                $config['subprocess'] ?? [],
                [
                    'enabled' => true,
                    'logger' => $config['logger'] ?? (string)$args->getOption('logger'),
                ],
            );

            $subprocessLogger = $this->getLogger($args, forceLogger: true);
            $processor = new SubprocessProcessor($subprocessLogger, $subprocessConfig, $this->container);
        } else {
            $processor = new $processorClass($logger, $this->container);
        }

        return $processor;
    }

    /**
     * @param \Cake\Console\Arguments $args Arguments
     * @param \Cake\Console\ConsoleIo $io ConsoleIo
     * @return int
     */
    public function execute(Arguments $args, ConsoleIo $io): int
    {
        $config = (string)$args->getOption('config');
        if (!Configure::check(sprintf('Queue.%s', $config))) {
            $io->error(sprintf('Configuration key "%s" was not found', $config));
            $this->abort();
        }

        $logger = $this->getLogger($args);
        $processor = $this->getProcessor($args, $io, $logger);
        $extension = $this->getQueueExtension($args, $logger);

        $hasListener = Configure::check(sprintf('Queue.%s.listener', $config));
        if ($hasListener) {
            $listenerClassName = Configure::read(sprintf('Queue.%s.listener', $config));
            if (!class_exists($listenerClassName)) {
                $io->error(sprintf('Listener class %s not found', $listenerClassName));
                $this->abort();
            }

            /** @var \Cake\Event\EventListenerInterface $listener */
            $listener = new $listenerClassName();

            if ($processor instanceof Processor) {
                $processor->getEventManager()->on($listener);
            }
        }

        $client = QueueManager::engine($config);
        $queue = $args->getOption('queue')
            ? (string)$args->getOption('queue')
            : Configure::read(sprintf('Queue.%s.queue', $config), 'default');
        $processorName = $args->getOption('processor') ? (string)$args->getOption('processor') : 'default';

        $client->bindTopic($queue, $processor, $processorName);
        $client->consume($extension);

        return self::CODE_SUCCESS;
    }
}