3 namespace Drupal\Core\Queue;
5 use Drupal\Core\Database\Connection;
6 use Drupal\Core\Database\SchemaObjectExistsException;
7 use Drupal\Core\DependencyInjection\DependencySerializationTrait;
10 * Default queue implementation.
14 class DatabaseQueue implements ReliableQueueInterface, QueueGarbageCollectionInterface {
16 use DependencySerializationTrait;
19 * The database table name.
21 const TABLE_NAME = 'queue';
24 * The name of the queue this instance is working with.
31 * The database connection.
33 * @var \Drupal\Core\Database\Connection
35 protected $connection;
38 * Constructs a \Drupal\Core\Queue\DatabaseQueue object.
41 * The name of the queue.
42 * @param \Drupal\Core\Database\Connection $connection
43 * The Connection object containing the key-value tables.
45 public function __construct($name, Connection $connection) {
47 $this->connection = $connection;
53 public function createItem($data) {
56 $id = $this->doCreateItem($data);
58 catch (\Exception $e) {
59 // If there was an exception, try to create the table.
60 if (!$try_again = $this->ensureTableExists()) {
61 // If the exception happened for other reason than the missing table,
62 // propagate the exception.
66 // Now that the table has been created, try again if necessary.
68 $id = $this->doCreateItem($data);
74 * Adds a queue item and store it directly to the queue.
77 * Arbitrary data to be associated with the new task in the queue.
80 * A unique ID if the item was successfully created and was (best effort)
81 * added to the queue, otherwise FALSE. We don't guarantee the item was
82 * committed to disk etc, but as far as we know, the item is now in the
85 protected function doCreateItem($data) {
86 $query = $this->connection->insert(static::TABLE_NAME)
88 'name' => $this->name,
89 'data' => serialize($data),
90 // We cannot rely on REQUEST_TIME because many items might be created
91 // by a single request which takes longer than 1 second.
94 // Return the new serial ID, or FALSE on failure.
95 return $query->execute();
101 public function numberOfItems() {
103 return $this->connection->query('SELECT COUNT(item_id) FROM {' . static::TABLE_NAME . '} WHERE name = :name', [':name' => $this->name])
106 catch (\Exception $e) {
107 $this->catchException($e);
108 // If there is no table there cannot be any items.
116 public function claimItem($lease_time = 30) {
117 // Claim an item by updating its expire fields. If claim is not successful
118 // another thread may have claimed the item in the meantime. Therefore loop
119 // until an item is successfully claimed or we are reasonably sure there
120 // are no unclaimed items left.
123 $item = $this->connection->queryRange('SELECT data, created, item_id FROM {' . static::TABLE_NAME . '} q WHERE expire = 0 AND name = :name ORDER BY created, item_id ASC', 0, 1, [':name' => $this->name])->fetchObject();
125 catch (\Exception $e) {
126 $this->catchException($e);
127 // If the table does not exist there are no items currently available to
132 // Try to update the item. Only one thread can succeed in UPDATEing the
133 // same row. We cannot rely on REQUEST_TIME because items might be
134 // claimed by a single consumer which runs longer than 1 second. If we
135 // continue to use REQUEST_TIME instead of the current time(), we steal
136 // time from the lease, and will tend to reset items before the lease
137 // should really expire.
138 $update = $this->connection->update(static::TABLE_NAME)
140 'expire' => time() + $lease_time,
142 ->condition('item_id', $item->item_id)
143 ->condition('expire', 0);
144 // If there are affected rows, this update succeeded.
145 if ($update->execute()) {
146 $item->data = unserialize($item->data);
151 // No items currently available to claim.
160 public function releaseItem($item) {
162 $update = $this->connection->update(static::TABLE_NAME)
166 ->condition('item_id', $item->item_id);
167 return $update->execute();
169 catch (\Exception $e) {
170 $this->catchException($e);
171 // If the table doesn't exist we should consider the item released.
179 public function deleteItem($item) {
181 $this->connection->delete(static::TABLE_NAME)
182 ->condition('item_id', $item->item_id)
185 catch (\Exception $e) {
186 $this->catchException($e);
193 public function createQueue() {
194 // All tasks are stored in a single database table (which is created on
195 // demand) so there is nothing we need to do to create a new queue.
201 public function deleteQueue() {
203 $this->connection->delete(static::TABLE_NAME)
204 ->condition('name', $this->name)
207 catch (\Exception $e) {
208 $this->catchException($e);
215 public function garbageCollection() {
217 // Clean up the queue for failed batches.
218 $this->connection->delete(static::TABLE_NAME)
219 ->condition('created', REQUEST_TIME - 864000, '<')
220 ->condition('name', 'drupal_batch:%', 'LIKE')
223 // Reset expired items in the default queue implementation table. If that's
224 // not used, this will simply be a no-op.
225 $this->connection->update(static::TABLE_NAME)
229 ->condition('expire', 0, '<>')
230 ->condition('expire', REQUEST_TIME, '<')
233 catch (\Exception $e) {
234 $this->catchException($e);
239 * Check if the table exists and create it if not.
241 protected function ensureTableExists() {
243 $database_schema = $this->connection->schema();
244 if (!$database_schema->tableExists(static::TABLE_NAME)) {
245 $schema_definition = $this->schemaDefinition();
246 $database_schema->createTable(static::TABLE_NAME, $schema_definition);
250 // If another process has already created the queue table, attempting to
251 // recreate it will throw an exception. In this case just catch the
252 // exception and do nothing.
253 catch (SchemaObjectExistsException $e) {
260 * Act on an exception when queue might be stale.
262 * If the table does not yet exist, that's fine, but if the table exists and
263 * yet the query failed, then the queue is stale and the exception needs to
270 * If the table exists the exception passed in is rethrown.
272 protected function catchException(\Exception $e) {
273 if ($this->connection->schema()->tableExists(static::TABLE_NAME)) {
279 * Defines the schema for the queue table.
283 public function schemaDefinition() {
285 'description' => 'Stores items in queues.',
291 'description' => 'Primary Key: Unique item ID.',
294 'type' => 'varchar_ascii',
298 'description' => 'The queue name.',
305 'description' => 'The arbitrary data for the item.',
311 'description' => 'Timestamp when the claim lease expires on the item.',
317 'description' => 'Timestamp when the item was created.',
320 'primary key' => ['item_id'],
322 'name_created' => ['name', 'created'],
323 'expire' => ['expire'],