DatabaseQueue.php 4.16 KB
Newer Older
1 2 3 4
<?php

/**
 * @file
5
 * Contains \Drupal\Core\Queue\DatabaseQueue.
6 7 8 9
 */

namespace Drupal\Core\Queue;

10 11
use Drupal\Core\Database\Connection;

12 13
/**
 * Default queue implementation.
14 15
 *
 * @ingroup queue
16
 */
17 18
class DatabaseQueue implements ReliableQueueInterface {

19 20 21 22 23 24 25 26
  /**
   * The name of the queue this instance is working with.
   *
   * @var string
   */
  protected $name;

  /**
27 28 29 30 31 32 33
   * The database connection.
   *
   * @var \Drupal\Core\Database\Connection $connection
   */
  protected $connection;

  /**
34
   * Constructs a \Drupal\Core\Queue\DatabaseQueue object.
35 36
   *
   * @param string $name
37 38 39
   *   The name of the queue.
   * @param \Drupal\Core\Database\Connection $connection
   *   The Connection object containing the key-value tables.
40
   */
41
  function __construct($name, Connection $connection) {
42
    $this->name = $name;
43
    $this->connection = $connection;
44 45 46 47 48 49
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::createItem().
   */
  public function createItem($data) {
50
    $query = $this->connection->insert('queue')
51 52 53 54 55 56 57
      ->fields(array(
        'name' => $this->name,
        'data' => serialize($data),
        // We cannot rely on REQUEST_TIME because many items might be created
        // by a single request which takes longer than 1 second.
        'created' => time(),
      ));
58 59
    // Return the new serial ID, or FALSE on failure.
    return $query->execute();
60 61 62 63 64 65
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::numberOfItems().
   */
  public function numberOfItems() {
66
    return $this->connection->query('SELECT COUNT(item_id) FROM {queue} WHERE name = :name', array(':name' => $this->name))->fetchField();
67 68 69 70 71 72 73 74 75 76 77
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::claimItem().
   */
  public function claimItem($lease_time = 30) {
    // Claim an item by updating its expire fields. If claim is not successful
    // another thread may have claimed the item in the meantime. Therefore loop
    // until an item is successfully claimed or we are reasonably sure there
    // are no unclaimed items left.
    while (TRUE) {
78
      $item = $this->connection->queryRange('SELECT data, created, item_id FROM {queue} q WHERE expire = 0 AND name = :name ORDER BY created ASC', 0, 1, array(':name' => $this->name))->fetchObject();
79 80 81 82 83 84 85
      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.
86
        $update = $this->connection->update('queue')
87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108
          ->fields(array(
            'expire' => time() + $lease_time,
          ))
          ->condition('item_id', $item->item_id)
          ->condition('expire', 0);
        // If there are affected rows, this update succeeded.
        if ($update->execute()) {
          $item->data = unserialize($item->data);
          return $item;
        }
      }
      else {
        // No items currently available to claim.
        return FALSE;
      }
    }
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::releaseItem().
   */
  public function releaseItem($item) {
109
    $update = $this->connection->update('queue')
110 111 112 113 114 115 116 117 118 119 120
      ->fields(array(
        'expire' => 0,
      ))
      ->condition('item_id', $item->item_id);
      return $update->execute();
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::deleteItem().
   */
  public function deleteItem($item) {
121
    $this->connection->delete('queue')
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138
      ->condition('item_id', $item->item_id)
      ->execute();
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::createQueue().
   */
  public function createQueue() {
    // All tasks are stored in a single database table (which is created when
    // Drupal is first installed) so there is nothing we need to do to create
    // a new queue.
  }

  /**
   * Implements Drupal\Core\Queue\QueueInterface::deleteQueue().
   */
  public function deleteQueue() {
139
    $this->connection->delete('queue')
140 141 142 143
      ->condition('name', $this->name)
      ->execute();
  }
}