Verified Commit 6b34cc92 authored by godotislate's avatar godotislate
Browse files

feat: #1189464 Add an 'instant' queue runner 

By: catch
By: berdir
By: tsphethean
By: sukr_s
By: smustgrave
By: heddn
By: nicxvan
By: godotislate
(cherry picked from commit 9a96116d)
parent 0757aebb

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
@@ -73,6 +73,8 @@ parameters: 
  tempstore.expire: 604800

  queue.config:

    suspendMaximumWait: 30.0

  instant_queue.config:

    maximumItems: 5

  stream_wrapper.allowed_file_extensions:

    module:

      - json
@@ -619,6 +621,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']
@@ -655,6 +660,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;

  }



}