Index: drupal_queue.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/drupal_queue/drupal_queue.inc,v
retrieving revision 1.3
diff -u -p -r1.3 drupal_queue.inc
--- drupal_queue.inc	31 Oct 2009 13:29:16 -0000	1.3
+++ drupal_queue.inc	31 Aug 2010 02:42:27 -0000
@@ -5,11 +5,7 @@
  * @file
  * Queue functionality.
  *
- * Copied from system.queue.inc.
- *
- * Note:
- * - Queries are rewritten
- * - REQUEST_TIME does not exist in D6, however, some comments refer to it.
+ * Backported from Drupal 7's system.queue.inc.
  */
 
 /**
@@ -28,29 +24,35 @@
  * 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
+ * made available again by the DrupalQueueInterface implementation once the
  * lease expires. Another consumer will then be able to receive it when calling
- * DrupalQueueInterface::claimItem().
+ * DrupalQueueInterface::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 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::claimItem() 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
+ * 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.
- *
- * 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.
+ * 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 can this not be a problem. 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).
  */
 
 /**
@@ -58,18 +60,38 @@
  */
 class DrupalQueue {
   /**
-   * Get a queue object for a given name.
+   * Returns the queue object for a given name.
+   *
+   * 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 SystemQueue, 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
+   *   SystemQueue.
    *
    * @param $name
    *   Arbitrary string. The name of the queue to work with.
+   * @param $reliable
+   *   TRUE if the ordering of items and guaranteeing every item executes at
+   *   least once is important, FALSE if scalability is the main concern.
+   *
    * @return
    *   The queue object for a given name.
    */
-  public static function get($name) {
+  public static function get($name, $reliable = FALSE) {
     static $queues;
     if (!isset($queues[$name])) {
-      $class = variable_get('queue_module_' . $name, 'System') . 'Queue';
-      $queues[$name] = new $class($name);
+      $class = variable_get('queue_class_' . $name, NULL);
+      if (!$class) {
+        $class = variable_get('queue_default_class', 'SystemQueue');
+      }
+      $object = new $class($name);
+      if ($reliable && !$object instanceof DrupalReliableQueueInterface) {
+        $class = variable_get('queue_default_reliable_class', 'SystemQueue');
+        $object = new $class($name);
+      }
+      $queues[$name] = $object;
     }
     return $queues[$name];
   }
@@ -92,8 +114,8 @@ interface DrupalQueueInterface {
    * @return
    *   TRUE 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, that your disk wasn't hit by a meteor, etc, but as
-   *   far as we know, the item is now in the queue.
+   *   committed to disk etc, but as far as we know, the item is now in the
+   *   queue.
    */
   public function createItem($data);
 
@@ -137,11 +159,20 @@ interface DrupalQueueInterface {
    * Delete a finished item from the queue.
    *
    * @param $item
-   *   The item returned by claimItem().
+   *   The item returned by DrupalQueueInterface::claimItem().
    */
   public function deleteItem($item);
 
   /**
+   * Release an item that the worker could not process, so another
+   * worker can come in and process it before the timeout expires.
+   *
+   * @param $item
+   * @return boolean
+   */
+  public function releaseItem($item);
+
+  /**
    * Create a queue.
    *
    * Called during installation and should be used to perform any necessary
@@ -161,10 +192,18 @@ interface DrupalQueueInterface {
 }
 
 /**
- * Default queue implementation.
+ * Reliable queue interface.
+ *
+ * Classes implementing this interface preserve the order of messages and
+ * guarantee that every item will be executed at least once.
  */
-class SystemQueue implements DrupalQueueInterface {
+interface DrupalReliableQueueInterface extends DrupalQueueInterface {
+}
 
+/**
+ * Default queue implementation.
+ */
+class SystemQueue implements DrupalReliableQueueInterface {
   /**
    * The name of the queue this instance is working with.
    *
@@ -177,13 +216,8 @@ class SystemQueue implements DrupalQueue
   }
 
   public function createItem($data) {
-    $record = new stdClass();
-    $record->name = $this->name;
-    $record->data = $data;
-    // We cannot rely on REQUEST_TIME because many items might be created by a
-    // single request which takes longer than 1 second.
-    $record->created = time();
-    return drupal_write_record('queue', $record) !== FALSE;
+    db_query("INSERT INTO {queue} (name, data, created) VALUES ('%s', '%s', %d)", $this->name, serialize($data), time());
+    return (bool) db_affected_rows();
   }
 
   public function numberOfItems() {
@@ -199,11 +233,7 @@ class SystemQueue implements DrupalQueue
       $item = db_fetch_object(db_query_range('SELECT data, item_id FROM {queue} q WHERE expire = 0 AND name = "%s" ORDER BY created ASC', $this->name, 0, 1));
       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
-        // claimed by a single consumer which runs longer than 1 second. If we
-        // 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.
+        // same row.
         db_query('UPDATE {queue} SET expire = %d WHERE item_id = %d AND expire = 0', time() + $lease_time, $item->item_id);
         // If there are affected rows, this update succeeded.
         if (db_affected_rows()) {
@@ -218,6 +248,11 @@ class SystemQueue implements DrupalQueue
     }
   }
 
+  public function releaseItem($item) {
+    db_query("UPDATE {queue} SET expire = 0 WHERE item_id = %d", $item->item_id);
+    return (bool) db_affected_rows();
+  }
+
   public function deleteItem($item) {
     db_query('DELETE FROM {queue} WHERE item_id = %d', $item->item_id);
   }
@@ -234,5 +269,78 @@ class SystemQueue implements DrupalQueue
 }
 
 /**
+ * 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".
  */
Index: drupal_queue.test
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/drupal_queue/drupal_queue.test,v
retrieving revision 1.1
diff -u -p -r1.1 drupal_queue.test
--- drupal_queue.test	31 Oct 2009 13:49:50 -0000	1.1
+++ drupal_queue.test	31 Aug 2010 02:42:27 -0000
@@ -21,6 +21,7 @@ class QueueTestCase extends DrupalWebTes
 
   public function setUp() {
     parent::setUp('drupal_queue');
+    drupal_queue_include();
   }
 
   /**
@@ -28,7 +29,6 @@ class QueueTestCase extends DrupalWebTes
    */
   function testQueue() {
     // Create two queues.
-    drupal_queue_include();
     $queue1 = DrupalQueue::get($this->randomName());
     $queue1->createQueue();
     $queue2 = DrupalQueue::get($this->randomName());
@@ -101,4 +101,4 @@ class QueueTestCase extends DrupalWebTes
     }
     return $score;
   }
-}
\ No newline at end of file
+}