function Cron::processQueues

Processes cron queues.

File

core/lib/Drupal/Core/Cron.php, line 130

Class

Cron

The Drupal core Cron service.

Namespace

Code

protected function processQueues() {
    $max_wait = (double) $this->queueConfig['suspendMaximumWait'];
    // Build a stack of queues to work on.
    
    /** @var array<array{process_from: int<0, max>, queue: \Drupal\Core\Queue\QueueInterface, worker: \Drupal\Core\Queue\QueueWorkerInterface}> $queues */
    $queues = [];
    foreach ($this->queueManager
        ->getDefinitions() as $queue_name => $queue_info) {
        if (!isset($queue_info['cron'])) {
            continue;
        }
        $queue = $this->queueFactory
            ->get($queue_name);
        // Make sure every queue exists. There is no harm in trying to recreate
        // an existing queue.
        $queue->createQueue();
        $worker = $this->queueManager
            ->createInstance($queue_name);
        $queues[] = [
            // Set process_from to zero so each queue is always processed
            // immediately for the first time. This process_from timestamp will
            // change if a queue throws a delayable SuspendQueueException.
'process_from' => 0,
            'queue' => $queue,
            'worker' => $worker,
        ];
    }
    // Work through stack of queues, re-adding to the stack when a delay is
    // necessary.
    while ($item = array_shift($queues)) {
        [
            'queue' => $queue,
            'worker' => $worker,
            'process_from' => $process_from,
        ] = $item;
        // Each queue will be processed immediately when it is reached for the
        // first time, as zero > currentTime will never be true.
        if ($process_from > $this->time
            ->getCurrentMicroTime()) {
            $this->usleep(round($process_from - $this->time
                ->getCurrentMicroTime(), 3) * 1000000);
        }
        try {
            $this->processQueue($queue, $worker);
        } catch (SuspendQueueException $e) {
            // Return to this queue after processing other queues if the delay is
            // within the threshold.
            if ($e->isDelayable() && $e->getDelay() < $max_wait) {
                $item['process_from'] = $this->time
                    ->getCurrentMicroTime() + $e->getDelay();
                // Place this queue back in the stack for processing later.
                array_push($queues, $item);
            }
        }
        // Reorder the queue by next 'process_from' timestamp.
        usort($queues, function (array $queueA, array $queueB) {
            return $queueA['process_from'] <=> $queueB['process_from'];
        });
    }
}