0

Final Class Yiisoft\Queue\QueueConsumer

All Classes | Methods
InheritanceYiisoft\Queue\QueueConsumer
ImplementsYiisoft\Queue\QueueConsumerInterface

Consumes messages for one logical queue.

Public Methods

Hide inherited methods

Method Description Defined By
__construct() Yiisoft\Queue\QueueConsumer
listen() Yiisoft\Queue\QueueConsumer
run() Yiisoft\Queue\QueueConsumer

Method Details

Hide inherited methods

__construct() public method

public mixed __construct ( Yiisoft\Queue\Worker\WorkerInterface $worker, Yiisoft\Queue\Cli\LoopInterface $loop, \Psr\Log\LoggerInterface $logger, ?\Yiisoft\Queue\Adapter\AdapterInterface $adapter = null, string|\BackedEnum $queueName = DefaultQueue::NAME )
$worker Yiisoft\Queue\Worker\WorkerInterface
$loop Yiisoft\Queue\Cli\LoopInterface
$logger \Psr\Log\LoggerInterface
$adapter ?\Yiisoft\Queue\Adapter\AdapterInterface
$queueName string|\BackedEnum 

                public function __construct(
    private readonly WorkerInterface $worker,
    private readonly LoopInterface $loop,
    private readonly LoggerInterface $logger,
    private readonly ?AdapterInterface $adapter = null,
    string|BackedEnum $queueName = DefaultQueue::NAME,
) {
    $this->queueName = StringNormalizer::normalize($queueName);
}
listen() public method

public void listen ( )

                public function listen(): void
{
    if ($this->adapter === null) {
        $this->logger->info('Cannot listen without an adapter. Queue is in synchronous mode.');
        return;
    }
    $this->logger->info('Start listening to the queue.');
    $this->adapter->subscribe(fn(MessageInterface $message): bool => $this->handle($message));
    $this->logger->info('Finish listening to the queue.');
}
run() public method

public integer run ( integer $max = 0 )
$max integer 

                public function run(int $max = 0): int
{
    if ($this->adapter === null) {
        $this->logger->debug('Queue is in synchronous mode (no adapter). Messages are processed on push. run() does nothing.');
        return 0;
    }
    $this->logger->debug('Start processing queue messages.');
    $count = 0;
    $this->adapter->runExisting(function (MessageInterface $message) use (&$count, $max): bool {
        if (($max > 0 && $count >= $max) || !$this->handle($message)) {
            return false;
        }
        $count++;
        return true;
    });
    $this->logger->info('Processed {count} queue messages.', ['count' => $count]);
    return $count;
}