feat: #1189464 Add an 'instant' queue runner (9a96116d) · Commits · project / drupal

core/core.services.yml

+10 −0

Changes for core/core.services.yml: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -68,6 +68,8 @@ parameters:
		tempstore.expire: 604800
		queue.config:
		suspendMaximumWait: 30.0
		instant_queue.config:
		maximumItems: 5
		stream_wrapper.allowed_file_extensions:
		module:
		- json
		@@ -614,6 +616,9 @@ services:
		logger.channel.menu:
		parent: logger.channel_base
		arguments: ['menu']
		logger.channel.queue:
		parent: logger.channel_base
		arguments: ['queue']
		logger.channel.router:
		parent: logger.channel_base
		arguments: ['router']
		@@ -650,6 +655,11 @@ services:
		path.current:
		class: Drupal\Core\Path\CurrentPathStack
		arguments: ['@request_stack']
		Drupal\Core\Queue\InstantQueueRunnerInterface:
		class: 'Drupal\Core\Queue\InstantQueueRunner'
		autowire: true
		tags:
		- { name: needs_destruction }
		Drupal\Core\Path\CurrentPathStack: '@path.current'
		http_handler_stack:
		class: GuzzleHttp\HandlerStack

core/lib/Drupal/Core/Cron.php

+6 −40

Changes for core/lib/Drupal/Core/Cron.php: 6 added lines, 40 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,13 +7,11 @@
		use Drupal\Component\Utility\Timer;
		use Drupal\Core\Extension\ModuleHandlerInterface;
		use Drupal\Core\Lock\LockBackendInterface;
		use Drupal\Core\Queue\DelayableQueueInterface;
		use Drupal\Core\Queue\DelayedRequeueException;
		use Drupal\Core\Queue\QueueFactory;
		use Drupal\Core\Queue\QueueInterface;
		use Drupal\Core\Queue\QueueProcessTrait;
		use Drupal\Core\Queue\QueueWorkerInterface;
		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;
		@@ -26,6 +24,7 @@
		* The Drupal core Cron service.
		*/
		class Cron implements CronInterface {
		use QueueProcessTrait;

		/**
		* The queue config.
		@@ -203,43 +202,10 @@ protected function processQueues() {
		protected function processQueue(QueueInterface $queue, QueueWorkerInterface $worker) {
		$lease_time = $worker->getPluginDefinition()['cron']['time'];
		$end = $this->time->getCurrentTime() + $lease_time;
		while ($this->time->getCurrentTime() < $end && ($item = $queue->claimItem($lease_time))) {
		try {
		$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) {
		// The worker requested the task be immediately requeued.
		$queue->releaseItem($item);
		}
		catch (SuspendQueueException $e) {
		// If the worker indicates the whole queue should be skipped, release
		// the item and go to the next queue.
		$queue->releaseItem($item);

		$this->logger->debug('A worker for @queue queue suspended further processing of the queue.', [
		'@queue' => $worker->getPluginId(),
		]);

		// Skip to the next queue.
		throw $e;
		}
		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.
		Error::logException($this->logger, $e);
		while ($this->time->getCurrentTime() < $end) {
		$claimed = $this->processItem($queue, $worker);
		if (!$claimed) {
		break;
		}
		}
		}

core/lib/Drupal/Core/Queue/InstantQueueRunner.php

0 → 100644

+68 −0

Changes for core/lib/Drupal/Core/Queue/InstantQueueRunner.php: 68 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		<?php

		declare(strict_types=1);

		namespace Drupal\Core\Queue;

		use Drupal\Core\DestructableInterface;
		use Drupal\Core\Logger\LoggerChannelInterface;
		use Symfony\Component\DependencyInjection\Attribute\Autowire;

		/**
		* Provides an instant queue runner that processes items in kernel.destruct.
		*/
		class InstantQueueRunner implements InstantQueueRunnerInterface, DestructableInterface {
		use QueueProcessTrait;

		/**
		* An array of integers keyed by queue name.
		*
		* Tracks the number of items registered per queue during the request.
		*/
		protected array $queues = [];

		public function __construct(
		protected QueueFactory $queueFactory,
		protected QueueWorkerManagerInterface $queueManager,
		#[Autowire(service: 'logger.channel.queue')]
		protected LoggerChannelInterface $logger,
		#[Autowire(param: 'instant_queue.config')]
		protected array $config,
		) {}

		/**
		* {@inheritdoc}
		*/
		public function registerQueue(string $queue_name): void {
		$this->queues[$queue_name] = ($this->queues[$queue_name] ?? 0) + 1;
		}

		/**
		* {@inheritdoc}
		*/
		public function destruct(): void {
		foreach ($this->queues as $queue_name => $count) {
		$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);
		$count = min($this->config['maximumItems'], $count);
		while ($count) {
		$count--;
		try {
		$claimed = $this->processItem($queue, $worker);
		if (!$claimed) {
		break;
		}
		}
		// If any item suspends the queue, stop processing this queue altogether
		// for this request.
		catch (SuspendQueueException) {
		break;
		}
		}
		}
		}

		}

core/lib/Drupal/Core/Queue/InstantQueueRunnerInterface.php

0 → 100644

+20 −0

Changes for core/lib/Drupal/Core/Queue/InstantQueueRunnerInterface.php: 20 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		<?php

		declare(strict_types=1);

		namespace Drupal\Core\Queue;

		/**
		* Provides an interface for adding queues to be run instantly.
		*/
		interface InstantQueueRunnerInterface {

		/**
		* Registers a queue to be run by the instant queue runner.
		*
		* @param string $queue_name
		* The queue name.
		*/
		public function registerQueue(string $queue_name): void;

		}

core/lib/Drupal/Core/Queue/QueueProcessTrait.php

0 → 100644

+74 −0

Changes for core/lib/Drupal/Core/Queue/QueueProcessTrait.php: 74 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		<?php

		declare(strict_types=1);

		namespace Drupal\Core\Queue;

		use Drupal\Core\Utility\Error;

		/**
		* Provides a trait for processing queue items.
		*/
		trait QueueProcessTrait {

		/**
		* Process a queue item.
		*
		* @param \Drupal\Core\Queue\QueueInterface $queue
		* The queue.
		* @param \Drupal\Core\Queue\QueueWorkerInterface $worker
		* The queue worker.
		*
		* @return bool
		* TRUE if an item was claimed, FALSE if not.
		*
		* @throws \Drupal\Core\Queue\SuspendQueueException
		* If the queue was suspended.
		*/
		protected function processItem(QueueInterface $queue, QueueWorkerInterface $worker): bool {
		$claimed = FALSE;
		$lease_time = $worker->getPluginDefinition()['cron']['time'] ?? NULL;
		if ($item = $lease_time === NULL ? $queue->claimItem() : $queue->claimItem($lease_time)) {
		$claimed = TRUE;
		try {
		$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) {
		// The worker requested the task be immediately requeued.
		$queue->releaseItem($item);
		}
		catch (SuspendQueueException $e) {
		// If the worker indicates the whole queue should be skipped, release
		// the item and go to the next queue.
		$queue->releaseItem($item);

		$this->logger->debug('A worker for @queue queue suspended further processing of the queue.', [
		'@queue' => $worker->getPluginId(),
		]);

		// Skip to the next queue.
		throw $e;
		}
		catch (\Throwable $e) {
		// In case of any other kind of exception, log it and leave the item
		// in the queue to be processed again later.
		Error::logException($this->logger, $e);
		}
		}
		return $claimed;
		}

		}