? drupal_queue-782616-port.patch
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	22 Jul 2010 16:19:15 -0000
@@ -30,7 +30,8 @@
  * 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().
+ * 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
@@ -39,18 +40,23 @@
  * 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
+ * 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,21 +64,41 @@
  */
 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];
-  }
+  }  
 }
 
 interface DrupalQueueInterface {
@@ -92,9 +118,9 @@ 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);
 
   /**
@@ -161,9 +187,18 @@ interface DrupalQueueInterface {
 }
 
 /**
+ * Reliable queue interface.
+ *
+ * Classes implementing this interface preserve the order of messages and
+ * guarantee that every item will be executed at least once.
+ */
+interface DrupalReliableQueueInterface extends DrupalQueueInterface {
+}
+
+/**
  * Default queue implementation.
  */
-class SystemQueue implements DrupalQueueInterface {
+class SystemQueue implements DrupalReliableQueueInterface {
 
   /**
    * The name of the queue this instance is working with.
@@ -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".
  */