danielebarchiesi@0: <?php
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * @file
danielebarchiesi@0:  * Queue functionality.
danielebarchiesi@0:  */
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * @defgroup queue Queue operations
danielebarchiesi@0:  * @{
danielebarchiesi@0:  * Queue items to allow later processing.
danielebarchiesi@0:  *
danielebarchiesi@0:  * The queue system allows placing items in a queue and processing them later.
danielebarchiesi@0:  * The system tries to ensure that only one consumer can process an item.
danielebarchiesi@0:  *
danielebarchiesi@0:  * Before a queue can be used it needs to be created by
danielebarchiesi@0:  * DrupalQueueInterface::createQueue().
danielebarchiesi@0:  *
danielebarchiesi@0:  * Items can be added to the queue by passing an arbitrary data object to
danielebarchiesi@0:  * DrupalQueueInterface::createItem().
danielebarchiesi@0:  *
danielebarchiesi@0:  * To process an item, call DrupalQueueInterface::claimItem() and specify how
danielebarchiesi@0:  * long you want to have a lease for working on that item. When finished
danielebarchiesi@0:  * processing, the item needs to be deleted by calling
danielebarchiesi@0:  * DrupalQueueInterface::deleteItem(). If the consumer dies, the item will be
danielebarchiesi@0:  * made available again by the DrupalQueueInterface implementation once the
danielebarchiesi@0:  * lease expires. Another consumer will then be able to receive it when calling
danielebarchiesi@0:  * DrupalQueueInterface::claimItem(). Due to this, the processing code should
danielebarchiesi@0:  * be aware that an item might be handed over for processing more than once.
danielebarchiesi@0:  *
danielebarchiesi@0:  * The $item object used by the DrupalQueueInterface can contain arbitrary
danielebarchiesi@0:  * metadata depending on the implementation. Systems using the interface should
danielebarchiesi@0:  * only rely on the data property which will contain the information passed to
danielebarchiesi@0:  * DrupalQueueInterface::createItem(). The full queue item returned by
danielebarchiesi@0:  * DrupalQueueInterface::claimItem() needs to be passed to
danielebarchiesi@0:  * DrupalQueueInterface::deleteItem() once processing is completed.
danielebarchiesi@0:  *
danielebarchiesi@0:  * There are two kinds of queue backends available: reliable, which preserves
danielebarchiesi@0:  * the order of messages and guarantees that every item will be executed at
danielebarchiesi@0:  * least once. The non-reliable kind only does a best effort to preserve order
danielebarchiesi@0:  * in messages and to execute them at least once but there is a small chance
danielebarchiesi@0:  * that some items get lost. For example, some distributed back-ends like
danielebarchiesi@0:  * Amazon SQS will be managing jobs for a large set of producers and consumers
danielebarchiesi@0:  * where a strict FIFO ordering will likely not be preserved. Another example
danielebarchiesi@0:  * would be an in-memory queue backend which might lose items if it crashes.
danielebarchiesi@0:  * However, such a backend would be able to deal with significantly more writes
danielebarchiesi@0:  * than a reliable queue and for many tasks this is more important. See
danielebarchiesi@0:  * aggregator_cron() for an example of how to effectively utilize a
danielebarchiesi@0:  * non-reliable queue. Another example is doing Twitter statistics -- the small
danielebarchiesi@0:  * possibility of losing a few items is insignificant next to power of the
danielebarchiesi@0:  * queue being able to keep up with writes. As described in the processing
danielebarchiesi@0:  * section, regardless of the queue being reliable or not, the processing code
danielebarchiesi@0:  * should be aware that an item might be handed over for processing more than
danielebarchiesi@0:  * once (because the processing code might time out before it finishes).
danielebarchiesi@0:  */
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * Factory class for interacting with queues.
danielebarchiesi@0:  */
danielebarchiesi@0: class DrupalQueue {
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Returns the queue object for a given name.
danielebarchiesi@0:    *
danielebarchiesi@0:    * The following variables can be set by variable_set or $conf overrides:
danielebarchiesi@0:    * - queue_class_$name: the class to be used for the queue $name.
danielebarchiesi@0:    * - queue_default_class: the class to use when queue_class_$name is not
danielebarchiesi@0:    *   defined. Defaults to SystemQueue, a reliable backend using SQL.
danielebarchiesi@0:    * - queue_default_reliable_class: the class to use when queue_class_$name is
danielebarchiesi@0:    *   not defined and the queue_default_class is not reliable. Defaults to
danielebarchiesi@0:    *   SystemQueue.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @param $name
danielebarchiesi@0:    *   Arbitrary string. The name of the queue to work with.
danielebarchiesi@0:    * @param $reliable
danielebarchiesi@0:    *   TRUE if the ordering of items and guaranteeing every item executes at
danielebarchiesi@0:    *   least once is important, FALSE if scalability is the main concern.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @return
danielebarchiesi@0:    *   The queue object for a given name.
danielebarchiesi@0:    */
danielebarchiesi@0:   public static function get($name, $reliable = FALSE) {
danielebarchiesi@0:     static $queues;
danielebarchiesi@0:     if (!isset($queues[$name])) {
danielebarchiesi@0:       $class = variable_get('queue_class_' . $name, NULL);
danielebarchiesi@0:       if (!$class) {
danielebarchiesi@0:         $class = variable_get('queue_default_class', 'SystemQueue');
danielebarchiesi@0:       }
danielebarchiesi@0:       $object = new $class($name);
danielebarchiesi@0:       if ($reliable && !$object instanceof DrupalReliableQueueInterface) {
danielebarchiesi@0:         $class = variable_get('queue_default_reliable_class', 'SystemQueue');
danielebarchiesi@0:         $object = new $class($name);
danielebarchiesi@0:       }
danielebarchiesi@0:       $queues[$name] = $object;
danielebarchiesi@0:     }
danielebarchiesi@0:     return $queues[$name];
danielebarchiesi@0:   }
danielebarchiesi@0: }
danielebarchiesi@0: 
danielebarchiesi@0: interface DrupalQueueInterface {
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Add a queue item and store it directly to the queue.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @param $data
danielebarchiesi@0:    *   Arbitrary data to be associated with the new task in the queue.
danielebarchiesi@0:    * @return
danielebarchiesi@0:    *   TRUE if the item was successfully created and was (best effort) added
danielebarchiesi@0:    *   to the queue, otherwise FALSE. We don't guarantee the item was
danielebarchiesi@0:    *   committed to disk etc, but as far as we know, the item is now in the
danielebarchiesi@0:    *   queue.
danielebarchiesi@0:    */
danielebarchiesi@0:   public function createItem($data);
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Retrieve the number of items in the queue.
danielebarchiesi@0:    *
danielebarchiesi@0:    * This is intended to provide a "best guess" count of the number of items in
danielebarchiesi@0:    * the queue. Depending on the implementation and the setup, the accuracy of
danielebarchiesi@0:    * the results of this function may vary.
danielebarchiesi@0:    *
danielebarchiesi@0:    * e.g. On a busy system with a large number of consumers and items, the
danielebarchiesi@0:    * result might only be valid for a fraction of a second and not provide an
danielebarchiesi@0:    * accurate representation.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @return
danielebarchiesi@0:    *   An integer estimate of the number of items in the queue.
danielebarchiesi@0:    */
danielebarchiesi@0:   public function numberOfItems();
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Claim an item in the queue for processing.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @param $lease_time
danielebarchiesi@0:    *   How long the processing is expected to take in seconds, defaults to an
danielebarchiesi@0:    *   hour. After this lease expires, the item will be reset and another
danielebarchiesi@0:    *   consumer can claim the item. For idempotent tasks (which can be run
danielebarchiesi@0:    *   multiple times without side effects), shorter lease times would result
danielebarchiesi@0:    *   in lower latency in case a consumer fails. For tasks that should not be
danielebarchiesi@0:    *   run more than once (non-idempotent), a larger lease time will make it
danielebarchiesi@0:    *   more rare for a given task to run multiple times in cases of failure,
danielebarchiesi@0:    *   at the cost of higher latency.
danielebarchiesi@0:    * @return
danielebarchiesi@0:    *   On success we return an item object. If the queue is unable to claim an
danielebarchiesi@0:    *   item it returns false. This implies a best effort to retrieve an item
danielebarchiesi@0:    *   and either the queue is empty or there is some other non-recoverable
danielebarchiesi@0:    *   problem.
danielebarchiesi@0:    */
danielebarchiesi@0:   public function claimItem($lease_time = 3600);
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Delete a finished item from the queue.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @param $item
danielebarchiesi@0:    *   The item returned by DrupalQueueInterface::claimItem().
danielebarchiesi@0:    */
danielebarchiesi@0:   public function deleteItem($item);
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Release an item that the worker could not process, so another
danielebarchiesi@0:    * worker can come in and process it before the timeout expires.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @param $item
danielebarchiesi@0:    * @return boolean
danielebarchiesi@0:    */
danielebarchiesi@0:   public function releaseItem($item);
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Create a queue.
danielebarchiesi@0:    *
danielebarchiesi@0:    * Called during installation and should be used to perform any necessary
danielebarchiesi@0:    * initialization operations. This should not be confused with the
danielebarchiesi@0:    * constructor for these objects, which is called every time an object is
danielebarchiesi@0:    * instantiated to operate on a queue. This operation is only needed the
danielebarchiesi@0:    * first time a given queue is going to be initialized (for example, to make
danielebarchiesi@0:    * a new database table or directory to hold tasks for the queue -- it
danielebarchiesi@0:    * depends on the queue implementation if this is necessary at all).
danielebarchiesi@0:    */
danielebarchiesi@0:   public function createQueue();
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Delete a queue and every item in the queue.
danielebarchiesi@0:    */
danielebarchiesi@0:   public function deleteQueue();
danielebarchiesi@0: }
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * Reliable queue interface.
danielebarchiesi@0:  *
danielebarchiesi@0:  * Classes implementing this interface preserve the order of messages and
danielebarchiesi@0:  * guarantee that every item will be executed at least once.
danielebarchiesi@0:  */
danielebarchiesi@0: interface DrupalReliableQueueInterface extends DrupalQueueInterface {
danielebarchiesi@0: }
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * Default queue implementation.
danielebarchiesi@0:  */
danielebarchiesi@0: class SystemQueue implements DrupalReliableQueueInterface {
danielebarchiesi@0:   /**
danielebarchiesi@0:    * The name of the queue this instance is working with.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @var string
danielebarchiesi@0:    */
danielebarchiesi@0:   protected $name;
danielebarchiesi@0: 
danielebarchiesi@0:   public function __construct($name) {
danielebarchiesi@0:     $this->name = $name;
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function createItem($data) {
danielebarchiesi@0:     // During a Drupal 6.x to 7.x update, drupal_get_schema() does not contain
danielebarchiesi@0:     // the queue table yet, so we cannot rely on drupal_write_record().
danielebarchiesi@0:     $query = db_insert('queue')
danielebarchiesi@0:       ->fields(array(
danielebarchiesi@0:         'name' => $this->name,
danielebarchiesi@0:         'data' => serialize($data),
danielebarchiesi@0:         // We cannot rely on REQUEST_TIME because many items might be created
danielebarchiesi@0:         // by a single request which takes longer than 1 second.
danielebarchiesi@0:         'created' => time(),
danielebarchiesi@0:       ));
danielebarchiesi@0:     return (bool) $query->execute();
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function numberOfItems() {
danielebarchiesi@0:     return db_query('SELECT COUNT(item_id) FROM {queue} WHERE name = :name', array(':name' => $this->name))->fetchField();
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function claimItem($lease_time = 30) {
danielebarchiesi@0:     // Claim an item by updating its expire fields. If claim is not successful
danielebarchiesi@0:     // another thread may have claimed the item in the meantime. Therefore loop
danielebarchiesi@0:     // until an item is successfully claimed or we are reasonably sure there
danielebarchiesi@0:     // are no unclaimed items left.
danielebarchiesi@0:     while (TRUE) {
danielebarchiesi@0:       $item = db_query_range('SELECT data, item_id FROM {queue} q WHERE expire = 0 AND name = :name ORDER BY created ASC', 0, 1, array(':name' => $this->name))->fetchObject();
danielebarchiesi@0:       if ($item) {
danielebarchiesi@0:         // Try to update the item. Only one thread can succeed in UPDATEing the
danielebarchiesi@0:         // same row. We cannot rely on REQUEST_TIME because items might be
danielebarchiesi@0:         // claimed by a single consumer which runs longer than 1 second. If we
danielebarchiesi@0:         // continue to use REQUEST_TIME instead of the current time(), we steal
danielebarchiesi@0:         // time from the lease, and will tend to reset items before the lease
danielebarchiesi@0:         // should really expire.
danielebarchiesi@0:         $update = db_update('queue')
danielebarchiesi@0:           ->fields(array(
danielebarchiesi@0:             'expire' => time() + $lease_time,
danielebarchiesi@0:           ))
danielebarchiesi@0:           ->condition('item_id', $item->item_id)
danielebarchiesi@0:           ->condition('expire', 0);
danielebarchiesi@0:         // If there are affected rows, this update succeeded.
danielebarchiesi@0:         if ($update->execute()) {
danielebarchiesi@0:           $item->data = unserialize($item->data);
danielebarchiesi@0:           return $item;
danielebarchiesi@0:         }
danielebarchiesi@0:       }
danielebarchiesi@0:       else {
danielebarchiesi@0:         // No items currently available to claim.
danielebarchiesi@0:         return FALSE;
danielebarchiesi@0:       }
danielebarchiesi@0:     }
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function releaseItem($item) {
danielebarchiesi@0:     $update = db_update('queue')
danielebarchiesi@0:       ->fields(array(
danielebarchiesi@0:         'expire' => 0,
danielebarchiesi@0:       ))
danielebarchiesi@0:       ->condition('item_id', $item->item_id);
danielebarchiesi@0:       return $update->execute();
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function deleteItem($item) {
danielebarchiesi@0:     db_delete('queue')
danielebarchiesi@0:       ->condition('item_id', $item->item_id)
danielebarchiesi@0:       ->execute();
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function createQueue() {
danielebarchiesi@0:     // All tasks are stored in a single database table (which is created when
danielebarchiesi@0:     // Drupal is first installed) so there is nothing we need to do to create
danielebarchiesi@0:     // a new queue.
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function deleteQueue() {
danielebarchiesi@0:     db_delete('queue')
danielebarchiesi@0:       ->condition('name', $this->name)
danielebarchiesi@0:       ->execute();
danielebarchiesi@0:   }
danielebarchiesi@0: }
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * Static queue implementation.
danielebarchiesi@0:  *
danielebarchiesi@0:  * This allows "undelayed" variants of processes relying on the Queue
danielebarchiesi@0:  * interface. The queue data resides in memory. It should only be used for
danielebarchiesi@0:  * items that will be queued and dequeued within a given page request.
danielebarchiesi@0:  */
danielebarchiesi@0: class MemoryQueue implements DrupalQueueInterface {
danielebarchiesi@0:   /**
danielebarchiesi@0:    * The queue data.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @var array
danielebarchiesi@0:    */
danielebarchiesi@0:   protected $queue;
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Counter for item ids.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @var int
danielebarchiesi@0:    */
danielebarchiesi@0:   protected $id_sequence;
danielebarchiesi@0: 
danielebarchiesi@0:   /**
danielebarchiesi@0:    * Start working with a queue.
danielebarchiesi@0:    *
danielebarchiesi@0:    * @param $name
danielebarchiesi@0:    *   Arbitrary string. The name of the queue to work with.
danielebarchiesi@0:    */
danielebarchiesi@0:   public function __construct($name) {
danielebarchiesi@0:     $this->queue = array();
danielebarchiesi@0:     $this->id_sequence = 0;
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function createItem($data) {
danielebarchiesi@0:     $item = new stdClass();
danielebarchiesi@0:     $item->item_id = $this->id_sequence++;
danielebarchiesi@0:     $item->data = $data;
danielebarchiesi@0:     $item->created = time();
danielebarchiesi@0:     $item->expire = 0;
danielebarchiesi@0:     $this->queue[$item->item_id] = $item;
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function numberOfItems() {
danielebarchiesi@0:     return count($this->queue);
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function claimItem($lease_time = 30) {
danielebarchiesi@0:     foreach ($this->queue as $key => $item) {
danielebarchiesi@0:       if ($item->expire == 0) {
danielebarchiesi@0:         $item->expire = time() + $lease_time;
danielebarchiesi@0:         $this->queue[$key] = $item;
danielebarchiesi@0:         return $item;
danielebarchiesi@0:       }
danielebarchiesi@0:     }
danielebarchiesi@0:     return FALSE;
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function deleteItem($item) {
danielebarchiesi@0:     unset($this->queue[$item->item_id]);
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function releaseItem($item) {
danielebarchiesi@0:     if (isset($this->queue[$item->item_id]) && $this->queue[$item->item_id]->expire != 0) {
danielebarchiesi@0:       $this->queue[$item->item_id]->expire = 0;
danielebarchiesi@0:       return TRUE;
danielebarchiesi@0:     }
danielebarchiesi@0:     return FALSE;
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function createQueue() {
danielebarchiesi@0:     // Nothing needed here.
danielebarchiesi@0:   }
danielebarchiesi@0: 
danielebarchiesi@0:   public function deleteQueue() {
danielebarchiesi@0:     $this->queue = array();
danielebarchiesi@0:     $this->id_sequence = 0;
danielebarchiesi@0:   }
danielebarchiesi@0: }
danielebarchiesi@0: 
danielebarchiesi@0: /**
danielebarchiesi@0:  * @} End of "defgroup queue".
danielebarchiesi@0:  */