function Cron::processQueues

Processes cron queues.

File

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

Class

Cron

The Drupal core Cron service.

Namespace

Code

protected function processQueues() {
    // Grab the defined cron queues.
    foreach ($this->queueManager
        ->getDefinitions() as $queue_name => $info) {
        if (isset($info['cron'])) {
            // Make sure every queue exists. There is no harm in trying to recreate
            // an existing queue.
            $this->queueFactory
                ->get($queue_name)
                ->createQueue();
            $queue_worker = $this->queueManager
                ->createInstance($queue_name);
            $end = $this->time
                ->getCurrentTime() + $info['cron']['time'];
            $queue = $this->queueFactory
                ->get($queue_name);
            $lease_time = $info['cron']['time'];
            while ($this->time
                ->getCurrentTime() < $end && ($item = $queue->claimItem($lease_time))) {
                try {
                    $queue_worker->processItem($item->data);
                    $queue->deleteItem($item);
                } catch (DelayedRequeueException $e) {
                    // The worker requested the task not be immediately re-queued.
                    // - If the queue doesn't support ::delayItem(), we should leave the
                    // item's current expiry time alone.
                    // - If the queue does support ::delayItem(), we should allow the
                    // queue to update the item's expiry using the requested delay.
                    if ($queue instanceof DelayableQueueInterface) {
                        // This queue can handle a custom delay; use the duration provided
                        // by the exception.
                        $queue->delayItem($item, $e->getDelay());
                    }
                } catch (RequeueException $e) {
                    // The worker requested the task be immediately requeued.
                    $queue->releaseItem($item);
                } catch (SuspendQueueException $e) {
                    // If the worker indicates there is a problem with the whole queue,
                    // release the item and skip to the next queue.
                    $queue->releaseItem($item);
                    watchdog_exception('cron', $e);
                    // Skip to the next queue.
                    continue 2;
                } catch (\Exception $e) {
                    // In case of any other kind of exception, log it and leave the item
                    // in the queue to be processed again later.
                    watchdog_exception('cron', $e);
                }
            }
        }
    }
}