Update to Drupal 8.1.0. For more information, see https://www.drupal.org/drupal-8.1.0-release-notes

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\Batch.
- */
-
 namespace Drupal\Core\Queue;
 
 /**
@@ -30,10 +25,15 @@ class Batch extends DatabaseQueue {
    * item to be claimed repeatedly until it is deleted.
    */
   public function claimItem($lease_time = 0) {
-    $item = $this->connection->queryRange('SELECT data, item_id FROM {queue} q WHERE name = :name ORDER BY item_id ASC', 0, 1, array(':name' => $this->name))->fetchObject();
-    if ($item) {
-      $item->data = unserialize($item->data);
-      return $item;
+    try {
+      $item = $this->connection->queryRange('SELECT data, item_id FROM {queue} q WHERE name = :name ORDER BY item_id ASC', 0, 1, array(':name' => $this->name))->fetchObject();
+      if ($item) {
+        $item->data = unserialize($item->data);
+        return $item;
+      }
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
     }
     return FALSE;
   }
@@ -49,9 +49,14 @@ class Batch extends DatabaseQueue {
    */
   public function getAllItems() {
     $result = array();
-    $items = $this->connection->query('SELECT data FROM {queue} q WHERE name = :name ORDER BY item_id ASC', array(':name' => $this->name))->fetchAll();
-    foreach ($items as $item) {
-      $result[] = unserialize($item->data);
+    try {
+      $items = $this->connection->query('SELECT data FROM {queue} q WHERE name = :name ORDER BY item_id ASC', array(':name' => $this->name))->fetchAll();
+      foreach ($items as $item) {
+        $result[] = unserialize($item->data);
+      }
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
     }
     return $result;
   }

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\BatchMemory.
- */
-
 namespace Drupal\Core\Queue;
 
 /**

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

@@ -1,13 +1,9 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\DatabaseQueue.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Core\Database\Connection;
+use Drupal\Core\Database\SchemaObjectExistsException;
 use Drupal\Core\DependencyInjection\DependencySerializationTrait;
 
 /**
@@ -15,10 +11,15 @@ use Drupal\Core\DependencyInjection\DependencySerializationTrait;
  *
  * @ingroup queue
  */
-class DatabaseQueue implements ReliableQueueInterface {
+class DatabaseQueue implements ReliableQueueInterface, QueueGarbageCollectionInterface {
 
   use DependencySerializationTrait;
 
+  /**
+   * The database table name.
+   */
+  const TABLE_NAME = 'queue';
+
   /**
    * The name of the queue this instance is working with.
    *
@@ -50,7 +51,39 @@ class DatabaseQueue implements ReliableQueueInterface {
    * {@inheritdoc}
    */
   public function createItem($data) {
-    $query = $this->connection->insert('queue')
+    $try_again = FALSE;
+    try {
+      $id = $this->doCreateItem($data);
+    }
+    catch (\Exception $e) {
+      // If there was an exception, try to create the table.
+      if (!$try_again = $this->ensureTableExists()) {
+        // If the exception happened for other reason than the missing table,
+        // propagate the exception.
+        throw $e;
+      }
+    }
+    // Now that the table has been created, try again if necessary.
+    if ($try_again) {
+      $id = $this->doCreateItem($data);
+    }
+    return $id;
+  }
+
+  /**
+   * Adds a queue item and store it directly to the queue.
+   *
+   * @param $data
+   *   Arbitrary data to be associated with the new task in the queue.
+   *
+   * @return
+   *   A unique ID if the item was successfully created and was (best effort)
+   *   added to the queue, otherwise FALSE. We don't guarantee the item was
+   *   committed to disk etc, but as far as we know, the item is now in the
+   *   queue.
+   */
+  protected function doCreateItem($data) {
+    $query = $this->connection->insert(static::TABLE_NAME)
       ->fields(array(
         'name' => $this->name,
         'data' => serialize($data),
@@ -66,7 +99,15 @@ class DatabaseQueue implements ReliableQueueInterface {
    * {@inheritdoc}
    */
   public function numberOfItems() {
-    return $this->connection->query('SELECT COUNT(item_id) FROM {queue} WHERE name = :name', array(':name' => $this->name))->fetchField();
+    try {
+      return $this->connection->query('SELECT COUNT(item_id) FROM {' . static::TABLE_NAME . '} WHERE name = :name', array(':name' => $this->name))
+        ->fetchField();
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
+      // If there is no table there cannot be any items.
+      return 0;
+    }
   }
 
   /**
@@ -78,7 +119,15 @@ class DatabaseQueue implements ReliableQueueInterface {
     // until an item is successfully claimed or we are reasonably sure there
     // are no unclaimed items left.
     while (TRUE) {
-      $item = $this->connection->queryRange('SELECT data, created, item_id FROM {queue} q WHERE expire = 0 AND name = :name ORDER BY created, item_id ASC', 0, 1, array(':name' => $this->name))->fetchObject();
+      try {
+        $item = $this->connection->queryRange('SELECT data, created, item_id FROM {' . static::TABLE_NAME . '} q WHERE expire = 0 AND name = :name ORDER BY created, item_id ASC', 0, 1, array(':name' => $this->name))->fetchObject();
+      }
+      catch (\Exception $e) {
+        $this->catchException($e);
+        // If the table does not exist there are no items currently available to
+        // claim.
+        return FALSE;
+      }
       if ($item) {
         // Try to update the item. Only one thread can succeed in UPDATEing the
         // same row. We cannot rely on REQUEST_TIME because items might be
@@ -86,7 +135,7 @@ class DatabaseQueue implements ReliableQueueInterface {
         // continue to use REQUEST_TIME instead of the current time(), we steal
         // time from the lease, and will tend to reset items before the lease
         // should really expire.
-        $update = $this->connection->update('queue')
+        $update = $this->connection->update(static::TABLE_NAME)
           ->fields(array(
             'expire' => time() + $lease_time,
           ))
@@ -109,38 +158,169 @@ class DatabaseQueue implements ReliableQueueInterface {
    * {@inheritdoc}
    */
   public function releaseItem($item) {
-    $update = $this->connection->update('queue')
-      ->fields(array(
-        'expire' => 0,
-      ))
-      ->condition('item_id', $item->item_id);
+    try {
+      $update = $this->connection->update(static::TABLE_NAME)
+        ->fields(array(
+          'expire' => 0,
+        ))
+        ->condition('item_id', $item->item_id);
       return $update->execute();
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
+      // If the table doesn't exist we should consider the item released.
+      return TRUE;
+    }
   }
 
   /**
    * {@inheritdoc}
    */
   public function deleteItem($item) {
-    $this->connection->delete('queue')
-      ->condition('item_id', $item->item_id)
-      ->execute();
+    try {
+      $this->connection->delete(static::TABLE_NAME)
+        ->condition('item_id', $item->item_id)
+        ->execute();
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
+    }
   }
 
   /**
    * {@inheritdoc}
    */
   public function createQueue() {
-    // All tasks are stored in a single database table (which is created when
-    // Drupal is first installed) so there is nothing we need to do to create
-    // a new queue.
+    // All tasks are stored in a single database table (which is created on
+    // demand) so there is nothing we need to do to create a new queue.
   }
 
   /**
    * {@inheritdoc}
    */
   public function deleteQueue() {
-    $this->connection->delete('queue')
-      ->condition('name', $this->name)
-      ->execute();
+    try {
+      $this->connection->delete(static::TABLE_NAME)
+        ->condition('name', $this->name)
+        ->execute();
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
+    }
   }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function garbageCollection() {
+    try {
+      // Clean up the queue for failed batches.
+      $this->connection->delete(static::TABLE_NAME)
+        ->condition('created', REQUEST_TIME - 864000, '<')
+        ->condition('name', 'drupal_batch:%', 'LIKE')
+        ->execute();
+
+      // Reset expired items in the default queue implementation table. If that's
+      // not used, this will simply be a no-op.
+      $this->connection->update(static::TABLE_NAME)
+        ->fields(array(
+          'expire' => 0,
+        ))
+        ->condition('expire', 0, '<>')
+        ->condition('expire', REQUEST_TIME, '<')
+        ->execute();
+    }
+    catch (\Exception $e) {
+      $this->catchException($e);
+    }
+  }
+
+  /**
+   * Check if the table exists and create it if not.
+   */
+  protected function ensureTableExists() {
+    try {
+      $database_schema = $this->connection->schema();
+      if (!$database_schema->tableExists(static::TABLE_NAME)) {
+        $schema_definition = $this->schemaDefinition();
+        $database_schema->createTable(static::TABLE_NAME, $schema_definition);
+        return TRUE;
+      }
+    }
+    // If another process has already created the queue table, attempting to
+    // recreate it will throw an exception. In this case just catch the
+    // exception and do nothing.
+    catch (SchemaObjectExistsException $e) {
+      return TRUE;
+    }
+    return FALSE;
+  }
+
+  /**
+   * Act on an exception when queue might be stale.
+   *
+   * If the table does not yet exist, that's fine, but if the table exists and
+   * yet the query failed, then the queue is stale and the exception needs to
+   * propagate.
+   *
+   * @param $e
+   *   The exception.
+   *
+   * @throws \Exception
+   *   If the table exists the exception passed in is rethrown.
+   */
+  protected function catchException(\Exception $e) {
+    if ($this->connection->schema()->tableExists(static::TABLE_NAME)) {
+      throw $e;
+    }
+  }
+
+  /**
+   * Defines the schema for the queue table.
+   */
+  public function schemaDefinition() {
+    return [
+      'description' => 'Stores items in queues.',
+      'fields' => [
+        'item_id' => [
+          'type' => 'serial',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'description' => 'Primary Key: Unique item ID.',
+        ],
+        'name' => [
+          'type' => 'varchar_ascii',
+          'length' => 255,
+          'not null' => TRUE,
+          'default' => '',
+          'description' => 'The queue name.',
+        ],
+        'data' => [
+          'type' => 'blob',
+          'not null' => FALSE,
+          'size' => 'big',
+          'serialize' => TRUE,
+          'description' => 'The arbitrary data for the item.',
+        ],
+        'expire' => [
+          'type' => 'int',
+          'not null' => TRUE,
+          'default' => 0,
+          'description' => 'Timestamp when the claim lease expires on the item.',
+        ],
+        'created' => [
+          'type' => 'int',
+          'not null' => TRUE,
+          'default' => 0,
+          'description' => 'Timestamp when the item was created.',
+        ],
+      ],
+      'primary key' => ['item_id'],
+      'indexes' => [
+        'name_created' => ['name', 'created'],
+        'expire' => ['expire'],
+      ],
+    ];
+  }
+
 }

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\Memory.
- */
-
 namespace Drupal\Core\Queue;
 
 /**

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueDatabaseFactory.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Core\Database\Connection;

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueFactory.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Core\Site\Settings;

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

@@ -0,0 +1,18 @@
+<?php
+
+namespace Drupal\Core\Queue;
+
+/**
+ * If the Drupal 'queue' service implements this interface, the
+ * garbageCollection() method will be called during cron.
+ *
+ * @see system_cron()
+ */
+interface QueueGarbageCollectionInterface {
+
+  /**
+   * Cleans queues of garbage.
+   */
+  public function garbageCollection();
+
+}

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueInterface.
- */
-
 namespace Drupal\Core\Queue;
 
 /**

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueWorkerBase.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Component\Plugin\PluginBase;

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueWorkerInterface.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Component\Plugin\PluginInspectionInterface;
@@ -26,6 +21,9 @@ interface QueueWorkerInterface extends PluginInspectionInterface {
    *   The data that was passed to
    *   \Drupal\Core\Queue\QueueInterface::createItem() when the item was queued.
    *
+   * @throws \Drupal\Core\Queue\RequeueException
+   *   Processing is not yet finished. This will allow another process to claim
+   *   the item immediately.
    * @throws \Exception
    *   A QueueWorker plugin may throw an exception to indicate there was a
    *   problem. The cron process will log the exception, and leave the item in

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueWorkerManager.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Core\Cache\CacheBackendInterface;

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\QueueWorkerManagerInterface.
- */
-
 namespace Drupal\Core\Queue;
 
 use Drupal\Component\Plugin\PluginManagerInterface;

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\ReliableQueueInterface.
- */
-
 namespace Drupal\Core\Queue;
 
 /**

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

@@ -0,0 +1,8 @@
+<?php
+
+namespace Drupal\Core\Queue;
+
+/**
+ * Throw this exception to release the item allowing it to be processed again.
+ */
+class RequeueException extends \RuntimeException {}

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

@@ -1,10 +1,5 @@
 <?php
 
-/**
- * @file
- * Contains \Drupal\Core\Queue\SuspendQueueException.
- */
-
 namespace Drupal\Core\Queue;
 
 /**