3 files changed, 93 insertions, 49 deletions

diff --git a/modules/system/system.install b/modules/system/system.install
index add4c37c4..519cbabda 100644
--- a/modules/system/system.install
+++ b/modules/system/system.install
@@ -496,7 +496,9 @@ function system_schema() {
     'fields' => array(
       'bid' => array(
         'description' => 'Primary Key: Unique batch ID.',
-        'type' => 'serial',
+        // This is not a serial column, to allow both progressive and
+        // non-progressive batches. See batch_process().
+        'type' => 'int',
         'unsigned' => TRUE,
         'not null' => TRUE,
       ),
@@ -2257,50 +2259,7 @@ function system_update_7021() {
  * Add the queue tables.
  */
 function system_update_7022() {
-  $schema['queue'] = array(
-    'description' => 'Stores items in queues.',
-    'fields' => array(
-      'item_id' => array(
-        'type' => 'serial',
-        'unsigned' => TRUE,
-        'not null' => TRUE,
-        'description' => 'Primary Key: Unique item ID.',
-      ),
-      'name' => array(
-        'type' => 'varchar',
-        'length' => 255,
-        'not null' => TRUE,
-        'default' => '',
-        'description' => 'The queue name.',
-      ),
-      'data' => array(
-        'type' => 'text',
-        'not null' => FALSE,
-        'size' => 'big',
-        'serialize' => TRUE,
-        'description' => 'The arbitrary data for the item.',
-      ),
-      'expire' => array(
-        'type' => 'int',
-        'not null' => TRUE,
-        'default' => 0,
-        'description' => 'Timestamp when the claim lease expires on the item.',
-      ),
-      'created' => array(
-        'type' => 'int',
-        'not null' => TRUE,
-        'default' => 0,
-        'description' => 'Timestamp when the item was created.',
-      ),
-    ),
-    'primary key' => array('item_id'),
-    'indexes' => array(
-      'name_created' => array('name', 'created'),
-      'expire' => array('expire'),
-    ),
-  );
-
-  db_create_table('queue', $schema['queue']);
+  // Moved to update_fix_d7_requirements().
 }
 
 /**
@@ -2728,6 +2687,13 @@ function system_update_7049() {
 }
 
 /**
+ * Change {batch}.id column from serial to regular int.
+ */
+function system_update_7050() {
+  db_change_field('batch', 'bid', 'bid', array('description' => 'Primary Key: Unique batch ID.', 'type' => 'int', 'unsigned' => TRUE, 'not null' => TRUE));
+}
+
+/**
  * @} End of "defgroup updates-6.x-to-7.x"
  * The next series of updates should start at 8000.
  */
diff --git a/modules/system/system.module b/modules/system/system.module
index e0da1720b..6247fcb2f 100644
--- a/modules/system/system.module
+++ b/modules/system/system.module
@@ -2695,10 +2695,6 @@ function system_cron() {
   db_delete('flood')
     ->condition('expiration', REQUEST_TIME, '<')
     ->execute();
-  // Cleanup the batch table.
-  db_delete('batch')
-    ->condition('timestamp', REQUEST_TIME - 864000, '<')
-    ->execute();
 
   // Remove temporary files that are older than DRUPAL_MAXIMUM_TEMP_FILE_AGE.
   // Use separate placeholders for the status to avoid a bug in some versions
@@ -2722,6 +2718,15 @@ function system_cron() {
     cache_clear_all(NULL, $table);
   }
 
+  // Cleanup the batch table and the queue for failed batches.
+  db_delete('batch')
+    ->condition('timestamp', REQUEST_TIME - 864000, '<')
+    ->execute();
+  db_delete('queue')
+    ->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.
   db_update('queue')
diff --git a/modules/system/system.queue.inc b/modules/system/system.queue.inc
index 5174a0a65..bac1ff26e 100644
--- a/modules/system/system.queue.inc
+++ b/modules/system/system.queue.inc
@@ -254,5 +254,78 @@ class SystemQueue implements DrupalQueueInterface {
 }
 
 /**
+ * Static queue implementation.
+ *
+ * This allows "undelayed" variants of processes relying on the Queue
+ * interface. The queue data resides in memory. It should only be used for
+ * items that will be queued and dequeued within a given page request.
+ */
+class MemoryQueue implements DrupalQueueInterface {
+  /**
+   * The queue data.
+   *
+   * @var array
+   */
+  protected $queue;
+
+  /**
+   * Counter for item ids.
+   *
+   * @var int
+   */
+  protected $id_sequence;
+
+  public function __construct($name) {
+    $this->queue = array();
+    $this->id_sequence = 0;
+  }
+
+  public function createItem($data) {
+    $item = new stdClass();
+    $item->item_id = $this->id_sequence++;
+    $item->data = $data;
+    $item->created = time();
+    $item->expire = 0;
+    $this->queue[$item->item_id] = $item;
+  }
+
+  public function numberOfItems() {
+    return count($this->queue);
+  }
+
+  public function claimItem($lease_time = 30) {
+    foreach ($this->queue as $key => $item) {
+      if ($item->expire == 0) {
+        $item->expire = time() + $lease_time;
+        $this->queue[$key] = $item;
+        return $item;
+      }
+    }
+    return FALSE;
+  }
+
+  public function deleteItem($item) {
+    unset($this->queue[$item->item_id]);
+  }
+
+  public function releaseItem($item) {
+    if (isset($this->queue[$item->item_id]) && $this->queue[$item->item_id]->expire != 0) {
+      $this->queue[$item->item_id]->expire = 0;
+      return TRUE;
+    }
+    return FALSE;
+  }
+
+  public function createQueue() {
+    // Nothing needed here.
+  }
+
+  public function deleteQueue() {
+    $this->queue = array();
+    $this->id_sequence = 0;
+  }
+}
+
+/**
  * @} End of "defgroup queue".
  */