- Patch #391340 by chx, dww, neclimdul, Crell, alex_b, et al: job queue API.

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 DrupalQueueInterface::createQueue().

Items can be added to the queue by passing an arbitrary data object to DrupalQueueInterface::createItem().

To process an item, call DrupalQueueInterface::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 DrupalQueueInterface::deleteItem(). If the consumer dies, the item will be made available again by the DrapalQueueInterface implementation once the lease expires. Another consumer will then be able to receive it when calling DrupalQueueInterface::claimItem().

The $item object used by the DrupalQueueInterface 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 DrupalQueueInterface::createItem(). The full queue item returned by DrupalQueueInterface::createItem() needs to be passed to DrupalQueueInterface::deleteItem() once processing is completed.

While the queue system makes a best effort to preserve order in messages, due to the pluggable nature of the queue, there is no guarantee that items will be delivered on claim in the order they were sent. For example, some implementations like beanstalkd or others with 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.

The system also makes no guarantees about a task only being executed once: callers that have non-idempotent tasks either need to live with the possiblity of the task being invoked multiple times in cases where a claim lease expires, or need to implement their own transactions to make their tasks idempotent.

1 files changed, 92 insertions, 0 deletions

diff --git a/modules/system/system.test b/modules/system/system.test
index 1e26330a9..60c872df0 100644
--- a/modules/system/system.test
+++ b/modules/system/system.test
@@ -895,3 +895,95 @@ class SystemThemeFunctionalTest extends DrupalWebTestCase {
     $this->assertRaw('themes/garland', t('Site default theme used on the add content page.'));
   }
 }
+
+
+/**
+ * Test the basic queue functionality.
+ */
+class QueueTestCase extends DrupalWebTestCase {
+  public static function getInfo() {
+    return array(
+      'name' => t('Queue functionality'),
+      'description' => t('Queues and dequeues a set of items to check the basic queue functionality.'),
+      'group' => t('System'),
+    );
+  }
+
+  /**
+   * Queues and dequeues a set of items to check the basic queue functionality.
+   */
+  function testQueue() {
+    // Create two queues.
+    $queue1 = DrupalQueue::get($this->randomName());
+    $queue1->createQueue();
+    $queue2 = DrupalQueue::get($this->randomName());
+    $queue2->createQueue();
+
+    // Create four items.
+    $data = array();
+    for ($i = 0; $i < 4; $i++) {
+      $data[] = array($this->randomName() => $this->randomName());
+    }
+
+    // Queue items 1 and 2 in the queue1.
+    $queue1->createItem($data[0]);
+    $queue1->createItem($data[1]);
+
+    // Retrieve two items from queue1.
+    $items = array();
+    $new_items = array();
+
+    $items[] = $item = $queue1->claimItem();
+    $new_items[] = $item->data;
+
+    $items[] = $item = $queue1->claimItem();
+    $new_items[] = $item->data;
+
+    // First two dequeued items should match the first two items we queued.
+    $this->assertEqual($this->queueScore($data, $new_items), 2, t('Two items matched'));
+
+    // Add two more items.
+    $queue1->createItem($data[2]);
+    $queue1->createItem($data[3]);
+
+    $this->assertTrue($queue1->numberOfItems(), t('Queue 1 is not empty after adding items.'));
+    $this->assertFalse($queue2->numberOfItems(), t('Queue 2 is empty while Queue 1 has items'));
+
+    $items[] = $item = $queue1->claimItem();
+    $new_items[] = $item->data;
+
+    $items[] = $item = $queue1->claimItem();
+    $new_items[] = $item->data;
+
+    // All dequeued items should match the items we queued exactly once,
+    // therefore the score must be exactly 4.
+    $this->assertEqual($this->queueScore($data, $new_items), 4, t('Four items matched'));
+
+    // There should be no duplicate items.
+    $this->assertEqual($this->queueScore($new_items, $new_items), 4, t('Four items matched'));
+
+    // Delete all items from queue1.
+    foreach ($items as $item) {
+      $queue1->deleteItem($item);
+    }
+
+    // Check that both queues are empty.
+    $this->assertFalse($queue1->numberOfItems(), t('Queue 1 is empty'));
+    $this->assertFalse($queue2->numberOfItems(), t('Queue 2 is empty'));
+  }
+
+  /**
+   * This function returns the number of equal items in two arrays.
+   */
+  function queueScore($items, $new_items) {
+    $score = 0;
+    foreach ($items as $item) {
+      foreach ($new_items as $new_item) {
+        if ($item === $new_item) {
+          $score++;
+        }
+      }
+    }
+    return $score;
+  }
+}