From ac0cee8f4577b0eacb48af6b45c7f9b2b30d7c1e Mon Sep 17 00:00:00 2001 From: Alex Pott Date: Tue, 23 Sep 2014 10:47:47 +0100 Subject: [PATCH] Issue #2038275 by tim.plunkett, mr.baileys, Sam Hermans: Convert hook_queue_info() to plugin system. --- core/core.services.yml | 5 +- .../Drupal/Core/Annotation/QueueWorker.php | 65 ++++++++++ core/lib/Drupal/Core/Cron.php | 22 +++- core/lib/Drupal/Core/Queue/QueueFactory.php | 2 +- .../lib/Drupal/Core/Queue/QueueWorkerBase.php | 22 ++++ .../Core/Queue/QueueWorkerInterface.php | 46 +++++++ .../Drupal/Core/Queue/QueueWorkerManager.php | 65 ++++++++++ .../Queue/QueueWorkerManagerInterface.php | 17 +++ core/modules/aggregator/aggregator.module | 17 --- .../Plugin/QueueWorker/AggregatorRefresh.php | 31 +++++ .../src/Tests/AggregatorCronTest.php | 10 +- core/modules/locale/locale.module | 74 +---------- .../Plugin/QueueWorker/LocaleTranslation.php | 122 ++++++++++++++++++ core/modules/system/system.api.php | 76 +---------- .../cron_queue_test/cron_queue_test.module | 38 ------ .../QueueWorker/CronQueueTestBrokenQueue.php | 32 +++++ .../QueueWorker/CronQueueTestException.php | 28 ++++ 17 files changed, 459 insertions(+), 213 deletions(-) create mode 100644 core/lib/Drupal/Core/Annotation/QueueWorker.php create mode 100644 core/lib/Drupal/Core/Queue/QueueWorkerBase.php create mode 100644 core/lib/Drupal/Core/Queue/QueueWorkerInterface.php create mode 100644 core/lib/Drupal/Core/Queue/QueueWorkerManager.php create mode 100644 core/lib/Drupal/Core/Queue/QueueWorkerManagerInterface.php create mode 100644 core/modules/aggregator/src/Plugin/QueueWorker/AggregatorRefresh.php create mode 100644 core/modules/locale/src/Plugin/QueueWorker/LocaleTranslation.php delete mode 100644 core/modules/system/tests/modules/cron_queue_test/cron_queue_test.module create mode 100644 core/modules/system/tests/modules/cron_queue_test/src/Plugin/QueueWorker/CronQueueTestBrokenQueue.php create mode 100644 core/modules/system/tests/modules/cron_queue_test/src/Plugin/QueueWorker/CronQueueTestException.php diff --git a/core/core.services.yml b/core/core.services.yml index d3f91ee028..f5ed5db9d7 100644 --- a/core/core.services.yml +++ b/core/core.services.yml @@ -162,7 +162,7 @@ services: arguments: ['@typed_data_manager'] cron: class: Drupal\Core\Cron - arguments: ['@module_handler', '@lock', '@queue', '@state', '@current_user', '@session_manager', '@logger.channel.cron'] + arguments: ['@module_handler', '@lock', '@queue', '@state', '@current_user', '@session_manager', '@logger.channel.cron', '@plugin.manager.queue_worker'] diff.formatter: class: Drupal\Core\Diff\DiffFormatter arguments: ['@config.factory'] @@ -353,6 +353,9 @@ services: plugin.manager.display_variant: class: Drupal\Core\Display\VariantManager parent: default_plugin_manager + plugin.manager.queue_worker: + class: Drupal\Core\Queue\QueueWorkerManager + parent: default_plugin_manager plugin.cache_clearer: class: Drupal\Core\Plugin\CachedDiscoveryClearer paramconverter.menu_link: diff --git a/core/lib/Drupal/Core/Annotation/QueueWorker.php b/core/lib/Drupal/Core/Annotation/QueueWorker.php new file mode 100644 index 0000000000..85a2296fd1 --- /dev/null +++ b/core/lib/Drupal/Core/Annotation/QueueWorker.php @@ -0,0 +1,65 @@ +moduleHandler = $module_handler; $this->lock = $lock; $this->queueFactory = $queue_factory; @@ -97,6 +107,7 @@ public function __construct(ModuleHandlerInterface $module_handler, LockBackendI $this->currentUser = $current_user; $this->sessionManager = $session_manager; $this->logger = $logger; + $this->queueManager = $queue_manager; } /** @@ -162,21 +173,18 @@ protected function setCronLastTime() { */ protected function processQueues() { // Grab the defined cron queues. - $queues = $this->moduleHandler->invokeAll('queue_info'); - $this->moduleHandler->alter('queue_info', $queues); - - foreach ($queues as $queue_name => $info) { + 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(); - $callback = $info['worker callback']; + $queue_worker = $this->queueManager->createInstance($queue_name); $end = time() + (isset($info['cron']['time']) ? $info['cron']['time'] : 15); $queue = $this->queueFactory->get($queue_name); while (time() < $end && ($item = $queue->claimItem())) { try { - call_user_func_array($callback, array($item->data)); + $queue_worker->processItem($item->data); $queue->deleteItem($item); } catch (SuspendQueueException $e) { diff --git a/core/lib/Drupal/Core/Queue/QueueFactory.php b/core/lib/Drupal/Core/Queue/QueueFactory.php index 6dce837991..70824c31c7 100644 --- a/core/lib/Drupal/Core/Queue/QueueFactory.php +++ b/core/lib/Drupal/Core/Queue/QueueFactory.php @@ -50,7 +50,7 @@ function __construct(Settings $settings) { * least once is important, FALSE if scalability is the main concern. Defaults * to FALSE. * - * @return \Drupal\Core\QueueStore\QueueInterface + * @return \Drupal\Core\Queue\QueueInterface * A queue implementation for the given name. */ public function get($name, $reliable = FALSE) { diff --git a/core/lib/Drupal/Core/Queue/QueueWorkerBase.php b/core/lib/Drupal/Core/Queue/QueueWorkerBase.php new file mode 100644 index 0000000000..7fe3417113 --- /dev/null +++ b/core/lib/Drupal/Core/Queue/QueueWorkerBase.php @@ -0,0 +1,22 @@ +setCacheBackend($cache_backend, 'queue_plugins'); + $this->alterInfo('queue_info'); + } + + /** + * {@inheritdoc} + */ + public function processDefinition(&$definition, $plugin_id) { + parent::processDefinition($definition, $plugin_id); + + // Assign a default time if a cron is specified. + if (isset($definition['cron'])) { + $definition['cron'] += [ + 'time' => 15, + ]; + } + } + + /** + * {@inheritdoc} + * + * @return \Drupal\Core\Queue\QueueWorkerInterface + */ + public function createInstance($plugin_id, array $configuration = []) { + return parent::createInstance($plugin_id, $configuration); + } + +} diff --git a/core/lib/Drupal/Core/Queue/QueueWorkerManagerInterface.php b/core/lib/Drupal/Core/Queue/QueueWorkerManagerInterface.php new file mode 100644 index 0000000000..e80776ba5d --- /dev/null +++ b/core/lib/Drupal/Core/Queue/QueueWorkerManagerInterface.php @@ -0,0 +1,17 @@ + t('Aggregator refresh'), - 'worker callback' => function (FeedInterface $feed) { - $feed->refreshItems(); - }, - 'cron' => array( - 'time' => 60, - ), - ); - return $queues; -} - /** * Renders the HTML content safely, as allowed. * diff --git a/core/modules/aggregator/src/Plugin/QueueWorker/AggregatorRefresh.php b/core/modules/aggregator/src/Plugin/QueueWorker/AggregatorRefresh.php new file mode 100644 index 0000000000..d3267eba42 --- /dev/null +++ b/core/modules/aggregator/src/Plugin/QueueWorker/AggregatorRefresh.php @@ -0,0 +1,31 @@ +refreshItems(); + } + } + +} diff --git a/core/modules/aggregator/src/Tests/AggregatorCronTest.php b/core/modules/aggregator/src/Tests/AggregatorCronTest.php index 8a8ee7d82f..59b6386108 100644 --- a/core/modules/aggregator/src/Tests/AggregatorCronTest.php +++ b/core/modules/aggregator/src/Tests/AggregatorCronTest.php @@ -21,11 +21,11 @@ public function testCron() { $this->createSampleNodes(); $feed = $this->createFeed(); $this->cronRun(); - $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField(), 'Expected number of items in database.'); + $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField()); $this->deleteFeedItems($feed); - $this->assertEqual(0, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField(), 'Expected number of items in database.'); + $this->assertEqual(0, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField()); $this->cronRun(); - $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField(), 'Expected number of items in database.'); + $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField()); // Test feed locking when queued for update. $this->deleteFeedItems($feed); @@ -36,7 +36,7 @@ public function testCron() { )) ->execute(); $this->cronRun(); - $this->assertEqual(0, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField(), 'Expected number of items in database.'); + $this->assertEqual(0, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField()); db_update('aggregator_feed') ->condition('fid', $feed->id()) ->fields(array( @@ -44,6 +44,6 @@ public function testCron() { )) ->execute(); $this->cronRun(); - $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField(), 'Expected number of items in database.'); + $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->id()))->fetchField()); } } diff --git a/core/modules/locale/locale.module b/core/modules/locale/locale.module index 5740e4a899..fe938a3a93 100644 --- a/core/modules/locale/locale.module +++ b/core/modules/locale/locale.module @@ -361,88 +361,18 @@ function locale_themes_uninstalled($themes) { /** * Implements hook_cron(). * - * @see locale_queue_info() + * @see \Drupal\locale\Plugin\QueueWorker\LocaleTranslation */ function locale_cron() { // Update translations only when an update frequency was set by the admin // and a translatable language was set. - // Update tasks are added to the queue here but processed by Drupal's cron - // using the cron worker defined in locale_queue_info(). + // Update tasks are added to the queue here but processed by Drupal's cron. if ($frequency = \Drupal::config('locale.settings')->get('translation.update_interval_days') && locale_translatable_language_list()) { module_load_include('translation.inc', 'locale'); locale_cron_fill_queue(); } } -/** - * Implements hook_queue_info(). - */ -function locale_queue_info() { - $queues['locale_translation'] = array( - 'title' => t('Update translations'), - 'worker callback' => 'locale_translation_worker', - 'cron' => array( - 'time' => 30, - ), - ); - return $queues; -} - -/** - * Callback: Executes interface translation queue tasks. - * - * The translation update functions executed here are batch operations which - * are also used in translation update batches. The batch functions may need to - * be executed multiple times to complete their task, typically this is the - * translation import function. When a batch function is not finished, a new - * queue task is created and added to the end of the queue. The batch context - * data is needed to continue the batch task is stored in the queue with the - * queue data. - * - * @param array $data - * Queue data array containing: - * - Function name. - * - Array of function arguments. Optionally contains the batch context data. - * - * @see locale_queue_info() - */ -function locale_translation_worker($data) { - module_load_include('batch.inc', 'locale'); - list($function, $args) = $data; - - // We execute batch operation functions here to check, download and import the - // translation files. Batch functions use a context variable as last argument - // which is passed by reference. When a batch operation is called for the - // first time a default batch context is created. When called iterative - // (usually the batch import function) the batch context is passed through via - // the queue and is part of the $data. - $last = count($args) - 1; - if (!is_array($args[$last]) || !isset($args[$last]['finished'])) { - $batch_context = array( - 'sandbox' => array(), - 'results' => array(), - 'finished' => 1, - 'message' => '', - ); - } - else { - $batch_context = $args[$last]; - unset ($args[$last]); - } - $args = array_merge($args, array(&$batch_context)); - - // Call the batch operation function. - call_user_func_array($function, $args); - - // If the batch operation is not finished we create a new queue task to - // continue the task. This is typically the translation import task. - if ($batch_context['finished'] < 1) { - unset($batch_context['strings']); - $queue = \Drupal::queue('locale_translation', TRUE); - $queue->createItem(array($function, $args)); - } -} - /** * Imports translations when new modules or themes are installed. * diff --git a/core/modules/locale/src/Plugin/QueueWorker/LocaleTranslation.php b/core/modules/locale/src/Plugin/QueueWorker/LocaleTranslation.php new file mode 100644 index 0000000000..a850b0d513 --- /dev/null +++ b/core/modules/locale/src/Plugin/QueueWorker/LocaleTranslation.php @@ -0,0 +1,122 @@ +moduleHandler = $module_handler; + $this->queue = $queue; + } + + /** + * {@inheritdoc} + */ + public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) { + return new static( + $configuration, + $plugin_id, + $plugin_definition, + $container->get('module_handler'), + $container->get('queue')->get('locale_translation', TRUE) + ); + } + + /** + * {@inheritdoc} + * + * The translation update functions executed here are batch operations which + * are also used in translation update batches. The batch functions may need + * to be executed multiple times to complete their task, typically this is the + * translation import function. When a batch function is not finished, a new + * queue task is created and added to the end of the queue. The batch context + * data is needed to continue the batch task is stored in the queue with the + * queue data. + */ + public function processItem($data) { + $this->moduleHandler->loadInclude('locale', 'batch.inc'); + list($function, $args) = $data; + + // We execute batch operation functions here to check, download and import + // the translation files. Batch functions use a context variable as last + // argument which is passed by reference. When a batch operation is called + // for the first time a default batch context is created. When called + // iterative (usually the batch import function) the batch context is passed + // through via the queue and is part of the $data. + $last = count($args) - 1; + if (!is_array($args[$last]) || !isset($args[$last]['finished'])) { + $batch_context = [ + 'sandbox' => [], + 'results' => [], + 'finished' => 1, + 'message' => '', + ]; + } + else { + $batch_context = $args[$last]; + unset ($args[$last]); + } + $args = array_merge($args, [&$batch_context]); + + // Call the batch operation function. + call_user_func_array($function, $args); + + // If the batch operation is not finished we create a new queue task to + // continue the task. This is typically the translation import task. + if ($batch_context['finished'] < 1) { + unset($batch_context['strings']); + $this->queue->createItem([$function, $args]); + } + } + +} diff --git a/core/modules/system/system.api.php b/core/modules/system/system.api.php index ae416ed556..3c3020f9b3 100644 --- a/core/modules/system/system.api.php +++ b/core/modules/system/system.api.php @@ -62,8 +62,8 @@ function hook_hook_info() { * Long-running tasks and tasks that could time out, such as retrieving remote * data, sending email, and intensive file tasks, should use the queue API * instead of executing the tasks directly. To do this, first define one or - * more queues via hook_queue_info(). Then, add items that need to be - * processed to the defined queues. + * more queues via a \Drupal\Core\Annotation\QueueWorker plugin. Then, add items + * that need to be processed to the defined queues. */ function hook_cron() { // Short-running operation example, not using a queue: @@ -98,46 +98,6 @@ function hook_data_type_info_alter(&$data_types) { $data_types['email']['class'] = '\Drupal\mymodule\Type\Email'; } -/** - * Declare queues holding items that need to be run periodically. - * - * While there can be only one hook_cron() process running at the same time, - * there can be any number of processes defined here running. Because of - * this, long running tasks are much better suited for this API. Items queued - * in hook_cron() might be processed in the same cron run if there are not many - * items in the queue, otherwise it might take several requests, which can be - * run in parallel. - * - * You can create queues, add items to them, claim them, etc without declaring - * the queue in this hook if you want, however, you need to take care of - * processing the items in the queue in that case. - * - * @return - * An associative array where the key is the queue name and the value is - * again an associative array. Possible keys are: - * - 'worker callback': A PHP callable to call that is an implementation of - * callback_queue_worker(). - * - 'cron': (optional) An associative array containing the optional key: - * - 'time': (optional) How much time Drupal cron should spend on calling - * this worker in seconds. Defaults to 15. - * If the cron key is not defined, the queue will not be processed by cron, - * and must be processed by other means. - * - * @see hook_cron() - * @see hook_queue_info_alter() - */ -function hook_queue_info() { - $queues['aggregator_feeds'] = array( - 'title' => t('Aggregator refresh'), - 'worker callback' => array('Drupal\my_module\MyClass', 'aggregatorRefresh'), - // Only needed if this queue should be processed by cron. - 'cron' => array( - 'time' => 60, - ), - ); - return $queues; -} - /** * Alter cron queue information before cron runs. * @@ -147,7 +107,8 @@ function hook_queue_info() { * @param array $queues * An array of cron queue information. * - * @see hook_queue_info() + * @see \Drupal\Core\QueueWorker\QueueWorkerInterface + * @see \Drupal\Core\Annotation\QueueWorker * @see \Drupal\Core\Cron */ function hook_queue_info_alter(&$queues) { @@ -156,35 +117,6 @@ function hook_queue_info_alter(&$queues) { $queues['aggregator_feeds']['cron']['time'] = 90; } -/** - * Work on a single queue item. - * - * Callback for hook_queue_info(). - * - * @param $queue_item_data - * The data that was passed to \Drupal\Core\Queue\QueueInterface::createItem() - * when the item was queued. - * - * @throws \Exception - * The worker callback may throw an exception to indicate there was a problem. - * The cron process will log the exception, and leave the item in the queue to - * be processed again later. - * @throws \Drupal\Core\Queue\SuspendQueueException - * More specifically, a SuspendQueueException should be thrown when the - * callback is aware that the problem will affect all subsequent workers of - * its queue. For example, a callback that makes HTTP requests may find that - * the remote server is not responding. The cron process will behave as with a - * normal Exception, and in addition will not attempt to process further items - * from the current item's queue during the current cron run. - * - * @see \Drupal\Core\Cron::run() - */ -function callback_queue_worker($queue_item_data) { - $node = node_load($queue_item_data); - $node->title = 'Updated title'; - $node->save(); -} - /** * Allows modules to declare their own Form API element types and specify their * default values. diff --git a/core/modules/system/tests/modules/cron_queue_test/cron_queue_test.module b/core/modules/system/tests/modules/cron_queue_test/cron_queue_test.module deleted file mode 100644 index c46fdb903c..0000000000 --- a/core/modules/system/tests/modules/cron_queue_test/cron_queue_test.module +++ /dev/null @@ -1,38 +0,0 @@ - t('Exception test'), - 'worker callback' => 'cron_queue_test_exception', - // Only needed if this queue should be processed by cron. - 'cron' => array( - 'time' => 60, - ), - ); - $queues['cron_queue_test_broken_queue'] = array( - 'title' => t('Broken queue test'), - 'worker callback' => 'cron_queue_test_broken_queue', - // Only needed if this queue should be processed by cron. - 'cron' => array( - 'time' => 60, - ), - ); - - return $queues; -} - -function cron_queue_test_exception($item) { - throw new Exception('That is not supposed to happen.'); -} - -/** - * Implements callback_queue_worker(). - * - * This queue is declared broken if the queue item data is 'crash'. - */ -function cron_queue_test_broken_queue($queue_item_data) { - if ($queue_item_data == 'crash') { - throw new \Drupal\Core\Queue\SuspendQueueException('The queue is broken.'); - } - // Do nothing otherwise. -} diff --git a/core/modules/system/tests/modules/cron_queue_test/src/Plugin/QueueWorker/CronQueueTestBrokenQueue.php b/core/modules/system/tests/modules/cron_queue_test/src/Plugin/QueueWorker/CronQueueTestBrokenQueue.php new file mode 100644 index 0000000000..f642a4d1bf --- /dev/null +++ b/core/modules/system/tests/modules/cron_queue_test/src/Plugin/QueueWorker/CronQueueTestBrokenQueue.php @@ -0,0 +1,32 @@ +