diff --git a/core/core.services.yml b/core/core.services.yml index b093d4bfac..cd24f8fd58 100644 --- a/core/core.services.yml +++ b/core/core.services.yml @@ -355,7 +355,7 @@ services: arguments: ['@service_container'] cron: class: Drupal\Core\Cron - arguments: ['@module_handler', '@lock', '@queue', '@state', '@account_switcher', '@logger.channel.cron', '@plugin.manager.queue_worker', '@datetime.time'] + arguments: ['@module_handler', '@lock', '@queue', '@state', '@account_switcher', '@logger.channel.cron', '@queue_processor', '@datetime.time'] lazy: true diff.formatter: class: Drupal\Core\Diff\DiffFormatter @@ -458,6 +458,9 @@ services: arguments: ['@settings'] calls: - [setContainer, ['@service_container']] + queue_processor: + class: Drupal\Core\QueueProcessor + arguments: ['@plugin.manager.queue_worker', '@queue', '@datetime.time'] queue.database: class: Drupal\Core\Queue\QueueDatabaseFactory arguments: ['@database'] diff --git a/core/lib/Drupal/Core/Cron.php b/core/lib/Drupal/Core/Cron.php index a18bf5b5a2..5100c00178 100644 --- a/core/lib/Drupal/Core/Cron.php +++ b/core/lib/Drupal/Core/Cron.php @@ -9,8 +9,6 @@ use Drupal\Core\Lock\LockBackendInterface; use Drupal\Core\Queue\QueueFactory; use Drupal\Core\Queue\QueueWorkerManagerInterface; -use Drupal\Core\Queue\RequeueException; -use Drupal\Core\Queue\SuspendQueueException; use Drupal\Core\Session\AccountSwitcherInterface; use Drupal\Core\Session\AnonymousUserSession; use Drupal\Core\State\StateInterface; @@ -36,13 +34,6 @@ class Cron implements CronInterface { */ protected $lock; - /** - * The queue service. - * - * @var \Drupal\Core\Queue\QueueFactory - */ - protected $queueFactory; - /** * The state service. * @@ -65,18 +56,18 @@ class Cron implements CronInterface { protected $logger; /** - * The queue plugin manager. + * The time service. * - * @var \Drupal\Core\Queue\QueueWorkerManagerInterface + * @var \Drupal\Component\Datetime\TimeInterface */ - protected $queueManager; + protected $time; /** - * The time service. + * The queue processor. * - * @var \Drupal\Component\Datetime\TimeInterface + * @var \Drupal\Core\QueueProcessor */ - protected $time; + private $queueProcessor; /** * Constructs a cron object. @@ -98,15 +89,14 @@ class Cron implements CronInterface { * @param \Drupal\Component\Datetime\TimeInterface $time * The time service. */ - public function __construct(ModuleHandlerInterface $module_handler, LockBackendInterface $lock, QueueFactory $queue_factory, StateInterface $state, AccountSwitcherInterface $account_switcher, LoggerInterface $logger, QueueWorkerManagerInterface $queue_manager, TimeInterface $time = NULL) { + public function __construct(ModuleHandlerInterface $module_handler, LockBackendInterface $lock, QueueFactory $queue_factory, StateInterface $state, AccountSwitcherInterface $account_switcher, LoggerInterface $logger, QueueProcessor $queue_processor, TimeInterface $time = NULL) { $this->moduleHandler = $module_handler; $this->lock = $lock; - $this->queueFactory = $queue_factory; $this->state = $state; $this->accountSwitcher = $account_switcher; $this->logger = $logger; - $this->queueManager = $queue_manager; $this->time = $time ?: \Drupal::service('datetime.time'); + $this->queueProcessor = $queue_processor; } /** @@ -142,7 +132,7 @@ public function run() { } // Process cron queues. - $this->processQueues(); + $this->queueProcessor->processCronQueues(); // Restore the user. $this->accountSwitcher->switchBack(); @@ -162,46 +152,13 @@ protected function setCronLastTime() { /** * Processes cron queues. + * + * @deprecated in Drupal 8.9.0 and is removed in Drupal 9.0.0. + * Instead, use the queue_processor service. */ 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 = time() + (isset($info['cron']['time']) ? $info['cron']['time'] : 15); - $queue = $this->queueFactory->get($queue_name); - $lease_time = isset($info['cron']['time']) ?: NULL; - while (time() < $end && ($item = $queue->claimItem($lease_time))) { - try { - $queue_worker->processItem($item->data); - $queue->deleteItem($item); - } - 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); - } - } - } - } + @trigger_error('Cron::processQueues is deprecated in Drupal 8.9.0 and is removed in Drupal 9.0.0. Instead, use the queue_processor service. See https://www.drupal.org/node/NEEDS-CHANGE-RECORD', E_USER_DEPRECATED); + $this->queueProcessor->processCronQueues(); } /** diff --git a/core/lib/Drupal/Core/Queue/QueueWorkerInterface.php b/core/lib/Drupal/Core/Queue/QueueWorkerInterface.php index fff1ae21cc..315fcadef0 100644 --- a/core/lib/Drupal/Core/Queue/QueueWorkerInterface.php +++ b/core/lib/Drupal/Core/Queue/QueueWorkerInterface.php @@ -37,7 +37,7 @@ interface QueueWorkerInterface extends PluginInspectionInterface { * process further items from the current item's queue during the current * cron run. * - * @see \Drupal\Core\Cron::processQueues() + * @see \Drupal\Core\QueueProcessor::processQueue() */ public function processItem($data); diff --git a/core/lib/Drupal/Core/QueueProcessor.php b/core/lib/Drupal/Core/QueueProcessor.php new file mode 100644 index 0000000000..72cf98a8c6 --- /dev/null +++ b/core/lib/Drupal/Core/QueueProcessor.php @@ -0,0 +1,111 @@ +queueManager = $queue_manager; + $this->queueFactory = $queue_factory; + $this->time = $time; + } + + /** + * Process the named queue. + * + * @param string $queue_name + * The name of the queue to process. + * @param int $process_time + * The amount of time to spend processing the queue, in seconds. + * @param int $lease_time + * The amount of time to lock each queue item for. + */ + public function processQueue(string $queue_name, int $process_time, int $lease_time) { + $queue_worker = $this->queueManager->createInstance($queue_name); + $queue = $this->queueFactory->get($queue_name); + $current_time = $this->time->getCurrentTime(); + while (($current_time < $current_time + $process_time) && ($item = $queue->claimItem($lease_time))) { + try { + $queue_worker->processItem($item->data); + $queue->deleteItem($item); + } + 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. + return; + } + 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); + } + + $current_time = $this->time->getCurrentTime(); + } + } + + /** + * Process all defined cron queues. + */ + public function processCronQueues() { + 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(); + + $time = isset($info['cron']['time']) ? $info['cron']['time'] : 15; + $lease_time = isset($info['cron']['time']) ?: NULL; + $this->processQueue($queue_name, $time, $lease_time); + } + } + } +}