diff --git a/core/includes/bootstrap.inc b/core/includes/bootstrap.inc index 6abe08a..4fcffb1 100644 --- a/core/includes/bootstrap.inc +++ b/core/includes/bootstrap.inc @@ -2329,6 +2329,9 @@ function drupal_container($reset = FALSE) { // functions. This default is overridden by drupal_language_initialize() // during language negotiation. $container->register(LANGUAGE_TYPE_INTERFACE, 'Drupal\\Core\\Language\\Language'); + + // Define the mappings to the Queue factory. + $container->register('queue', 'Drupal\\Core\\Queue\\QueueFactory'); } return $container; } diff --git a/core/includes/common.inc b/core/includes/common.inc index 353a9b5..f892c10 100644 --- a/core/includes/common.inc +++ b/core/includes/common.inc @@ -5324,7 +5324,7 @@ function drupal_cron_run() { // existing queue. foreach ($queues as $queue_name => $info) { if (isset($info['cron'])) { - queue($queue_name)->createQueue(); + drupal_container()->get('queue')->queue($queue_name)->createQueue(); } } // Register shutdown callback. @@ -5356,7 +5356,7 @@ function drupal_cron_run() { if (isset($info['cron'])) { $function = $info['worker callback']; $end = time() + (isset($info['cron']['time']) ? $info['cron']['time'] : 15); - $queue = queue($queue_name); + $queue = drupal_container()->get('queue')->queue($queue_name); while (time() < $end && ($item = $queue->claimItem())) { $function($item->data); $queue->deleteItem($item); @@ -7610,97 +7610,3 @@ function drupal_get_filetransfer_info() { } return $info; } - -/** - * @defgroup queue Queue operations - * @{ - * Queue items to allow later processing. - * - * The queue system allows placing items in a queue and processing them later. - * The system tries to ensure that only one consumer can process an item. - * - * Before a queue can be used it needs to be created by - * Drupal\Core\Queue\QueueInterface::createQueue(). - * - * Items can be added to the queue by passing an arbitrary data object to - * Drupal\Core\Queue\QueueInterface::createItem(). - * - * To process an item, call Drupal\Core\Queue\QueueInterface::claimItem() and - * specify how long you want to have a lease for working on that item. - * When finished processing, the item needs to be deleted by calling - * Drupal\Core\Queue\QueueInterface::deleteItem(). If the consumer dies, the - * item will be made available again by the Drupal\Core\Queue\QueueInterface - * implementation once the lease expires. Another consumer will then be able to - * receive it when calling Drupal\Core\Queue\QueueInterface::claimItem(). - * Due to this, the processing code should be aware that an item might be handed - * over for processing more than once. - * - * The $item object used by the Drupal\Core\Queue\QueueInterface can contain - * arbitrary metadata depending on the implementation. Systems using the - * interface should only rely on the data property which will contain the - * information passed to Drupal\Core\Queue\QueueInterface::createItem(). - * The full queue item returned by Drupal\Core\Queue\QueueInterface::claimItem() - * needs to be passed to Drupal\Core\Queue\QueueInterface::deleteItem() once - * processing is completed. - * - * There are two kinds of queue backends available: reliable, which preserves - * the order of messages and guarantees that every item will be executed at - * least once. The non-reliable kind only does a best effort to preserve order - * in messages and to execute them at least once but there is a small chance - * that some items get lost. For example, some distributed back-ends like - * Amazon SQS will be managing jobs for a large set of producers and consumers - * where a strict FIFO ordering will likely not be preserved. Another example - * would be an in-memory queue backend which might lose items if it crashes. - * However, such a backend would be able to deal with significantly more writes - * than a reliable queue and for many tasks this is more important. See - * aggregator_cron() for an example of how to effectively utilize a - * non-reliable queue. Another example is doing Twitter statistics -- the small - * possibility of losing a few items is insignificant next to power of the - * queue being able to keep up with writes. As described in the processing - * section, regardless of the queue being reliable or not, the processing code - * should be aware that an item might be handed over for processing more than - * once (because the processing code might time out before it finishes). - */ - -/** - * Instantiates and statically caches the correct class for a queue. - * - * The following variables can be set by variable_set or $conf overrides: - * - queue_class_$name: the class to be used for the queue $name. - * - queue_default_class: the class to use when queue_class_$name is not - * defined. Defaults to Drupal\Core\Queue\System, a reliable backend using - * SQL. - * - queue_default_reliable_class: the class to use when queue_class_$name is - * not defined and the queue_default_class is not reliable. Defaults to - * Drupal\Core\Queue\System. - * - * @param string $name - * The name of the queue to work with. - * @param bool $reliable - * TRUE if the ordering of items and guaranteeing every item executes at - * least once is important, FALSE if scalability is the main concern. Defaults - * to FALSE. - * - * @return Drupal\Core\Queue\QueueInterface - * The queue object for a given name. - * - * @see Drupal\Core\Queue\QueueInterface - */ -function queue($name, $reliable = FALSE) { - static $queues; - if (!isset($queues[$name])) { - $class = variable_get('queue_class_' . $name, NULL); - if ($class && $reliable && in_array('Drupal\Core\Queue\ReliableQueueInterface', class_implements($class))) { - $class = variable_get('queue_default_reliable_class', 'Drupal\Core\Queue\System'); - } - elseif (!$class) { - $class = variable_get('queue_default_class', 'Drupal\Core\Queue\System'); - } - $queues[$name] = new $class($name); - } - return $queues[$name]; -} - -/** - * @} End of "defgroup queue". - */ diff --git a/core/lib/Drupal/Core/Queue/QueueFactory.php b/core/lib/Drupal/Core/Queue/QueueFactory.php new file mode 100644 index 0000000..4b24856 --- /dev/null +++ b/core/lib/Drupal/Core/Queue/QueueFactory.php @@ -0,0 +1,101 @@ +queues[$name])) { + $class = variable_get('queue_class_' . $name, NULL); + if ($class && $reliable && in_array('Drupal\Core\Queue\ReliableQueueInterface', class_implements($class))) { + $class = variable_get('queue_default_reliable_class', 'Drupal\Core\Queue\System'); + } + elseif (!$class) { + $class = variable_get('queue_default_class', 'Drupal\Core\Queue\System'); + } + $this->queues[$name] = new $class($name); + } + return $this->queues[$name]; + } +} diff --git a/core/modules/aggregator/aggregator.module b/core/modules/aggregator/aggregator.module index caac279..51e58a5 100644 --- a/core/modules/aggregator/aggregator.module +++ b/core/modules/aggregator/aggregator.module @@ -312,7 +312,7 @@ function aggregator_cron() { ':time' => REQUEST_TIME, ':never' => AGGREGATOR_CLEAR_NEVER )); - $queue = queue('aggregator_feeds'); + $queue = drupal_container()->get('queue')->queue('aggregator_feeds'); foreach ($result as $feed) { if ($queue->createItem($feed)) { // Add timestamp to avoid queueing item more than once. diff --git a/core/modules/system/system.api.php b/core/modules/system/system.api.php index c6e1288..26ad777 100644 --- a/core/modules/system/system.api.php +++ b/core/modules/system/system.api.php @@ -138,7 +138,7 @@ function hook_cron() { ':time' => REQUEST_TIME, ':never' => AGGREGATOR_CLEAR_NEVER, )); - $queue = queue('aggregator_feeds'); + $queue = drupal_container()->get('queue')->queue('aggregator_feeds'); foreach ($result as $feed) { $queue->createItem($feed); } diff --git a/core/modules/update/update.fetch.inc b/core/modules/update/update.fetch.inc index 1b4e6ca..d44ac83 100644 --- a/core/modules/update/update.fetch.inc +++ b/core/modules/update/update.fetch.inc @@ -28,7 +28,7 @@ function update_manual_status() { * Process a step in the batch for fetching available update data. */ function update_fetch_data_batch(&$context) { - $queue = queue('update_fetch_tasks'); + $queue = drupal_container()->get('queue')->queue('update_fetch_tasks'); if (empty($context['sandbox']['max'])) { $context['finished'] = 0; $context['sandbox']['max'] = $queue->numberOfItems(); @@ -99,7 +99,7 @@ function update_fetch_data_finished($success, $results) { * Attempt to drain the queue of tasks for release history data to fetch. */ function _update_fetch_data() { - $queue = queue('update_fetch_tasks'); + $queue = drupal_container()->get('queue')->queue('update_fetch_tasks'); $end = time() + variable_get('update_max_fetch_time', UPDATE_MAX_FETCH_TIME); while (time() < $end && ($item = $queue->claimItem())) { _update_process_fetch_task($item->data); @@ -235,7 +235,7 @@ function _update_create_fetch_task($project) { } $cid = 'fetch_task::' . $project['name']; if (empty($fetch_tasks[$cid])) { - $queue = queue('update_fetch_tasks'); + $queue = drupal_container()->get('queue')->queue('update_fetch_tasks'); $queue->createItem($project); // Due to race conditions, it is possible that another process already // inserted a row into the {cache_update} table and the following query will diff --git a/core/modules/update/update.install b/core/modules/update/update.install index 9ff39d1..9a259cf 100644 --- a/core/modules/update/update.install +++ b/core/modules/update/update.install @@ -70,7 +70,7 @@ function update_schema() { * Implements hook_install(). */ function update_install() { - $queue = queue('update_fetch_tasks', TRUE); + $queue = drupal_container()->get('queue')->queue('update_fetch_tasks', TRUE); $queue->createQueue(); } @@ -91,7 +91,7 @@ function update_uninstall() { foreach ($variables as $variable) { variable_del($variable); } - $queue = queue('update_fetch_tasks'); + $queue = drupal_container()->get('queue')->queue('update_fetch_tasks'); $queue->deleteQueue(); } diff --git a/core/modules/update/update.test b/core/modules/update/update.test index c8cfdd4..6e4329a 100644 --- a/core/modules/update/update.test +++ b/core/modules/update/update.test @@ -235,7 +235,7 @@ class UpdateCoreTestCase extends UpdateTestHelper { $projectb = array( 'name' => 'bbb_update_test', ); - $queue = queue('update_fetch_tasks'); + $queue = drupal_container()->get('queue')->queue('update_fetch_tasks'); $this->assertEqual($queue->numberOfItems(), 0, 'Queue is empty'); update_create_fetch_task($projecta); $this->assertEqual($queue->numberOfItems(), 1, 'Queue contains one item');