· 4 years ago · Jun 12, 2021, 07:24 PM
1<?php
2
3namespace Drupal\Core\Queue;
4
5use Drupal\Core\Database\Connection;
6use Drupal\Core\Database\DatabaseException;
7use Drupal\Core\DependencyInjection\DependencySerializationTrait;
8
9/**
10 * Default queue implementation.
11 *
12 * @ingroup queue
13 */
14class DatabaseQueue implements ReliableQueueInterface, QueueGarbageCollectionInterface, DelayableQueueInterface {
15
16 use DependencySerializationTrait;
17
18 /**
19 * The database table name.
20 */
21 const TABLE_NAME = 'queue';
22
23 /**
24 * The name of the queue this instance is working with.
25 *
26 * @var string
27 */
28 protected $name;
29
30 /**
31 * The database connection.
32 *
33 * @var \Drupal\Core\Database\Connection
34 */
35 protected $connection;
36
37 /**
38 * Constructs a \Drupal\Core\Queue\DatabaseQueue object.
39 *
40 * @param string $name
41 * The name of the queue.
42 * @param \Drupal\Core\Database\Connection $connection
43 * The Connection object containing the key-value tables.
44 */
45 public function __construct($name, Connection $connection) {
46 $this->name = $name;
47 $this->connection = $connection;
48 }
49
50 /**
51 * {@inheritdoc}
52 */
53 public function createItem($data) {
54 $try_again = FALSE;
55 try {
56 $id = $this->doCreateItem($data);
57 }
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.
63 throw $e;
64 }
65 }
66 // Now that the table has been created, try again if necessary.
67 if ($try_again) {
68 $id = $this->doCreateItem($data);
69 }
70 return $id;
71 }
72
73 /**
74 * Adds a queue item and store it directly to the queue.
75 *
76 * @param $data
77 * Arbitrary data to be associated with the new task in the queue.
78 *
79 * @return
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
83 * queue.
84 */
85 protected function doCreateItem($data) {
86 $query = $this->connection->insert(static::TABLE_NAME)
87 ->fields([
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.
92 'created' => \Drupal::time()->getCurrentTime(),
93 ]);
94 // Return the new serial ID, or FALSE on failure.
95 return $query->execute();
96 }
97
98 /**
99 * {@inheritdoc}
100 */
101 public function numberOfItems() {
102 try {
103 return (int) $this->connection->query('SELECT COUNT([item_id]) FROM {' . static::TABLE_NAME . '} WHERE [name] = :name', [':name' => $this->name])
104 ->fetchField();
105 }
106 catch (\Exception $e) {
107 $this->catchException($e);
108 // If there is no table there cannot be any items.
109 return 0;
110 }
111 }
112
113 /**
114 * {@inheritdoc}
115 */
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.
121 while (TRUE) {
122 try {
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();
124 }
125 catch (\Exception $e) {
126 $this->catchException($e);
127 }
128
129 // If the table does not exist there are no items currently available to
130 // claim.
131 if (empty($item)) {
132 return FALSE;
133 }
134
135 // Try to update the item. Only one thread can succeed in UPDATEing the
136 // same row. We cannot rely on REQUEST_TIME because items might be
137 // claimed by a single consumer which runs longer than 1 second. If we
138 // continue to use REQUEST_TIME instead of the current time(), we steal
139 // time from the lease, and will tend to reset items before the lease
140 // should really expire.
141 $update = $this->connection->update(static::TABLE_NAME)
142 ->fields([
143 'expire' => \Drupal::time()->getCurrentTime() + $lease_time,
144 ])
145 ->condition('item_id', $item->item_id)
146 ->condition('expire', 0);
147 // If there are affected rows, this update succeeded.
148 if ($update->execute()) {
149 $item->data = unserialize($item->data);
150 return $item;
151 }
152 }
153 }
154
155 /**
156 * {@inheritdoc}
157 */
158 public function releaseItem($item) {
159 try {
160 $update = $this->connection->update(static::TABLE_NAME)
161 ->fields([
162 'expire' => 0,
163 ])
164 ->condition('item_id', $item->item_id);
165 return (bool) $update->execute();
166 }
167 catch (\Exception $e) {
168 $this->catchException($e);
169 // If the table doesn't exist we should consider the item released.
170 return TRUE;
171 }
172 }
173
174 /**
175 * {@inheritdoc}
176 */
177 public function delayItem($item, int $delay) {
178 // Only allow a positive delay interval.
179 if ($delay < 0) {
180 throw new \InvalidArgumentException('$delay must be non-negative');
181 }
182
183 try {
184 // Add the delay relative to the current time.
185 $expire = \Drupal::time()->getCurrentTime() + $delay;
186 // Update the expiry time of this item.
187 $update = $this->connection->update(static::TABLE_NAME)
188 ->fields([
189 'expire' => $expire,
190 ])
191 ->condition('item_id', $item->item_id);
192 return (bool) $update->execute();
193 }
194 catch (\Exception $e) {
195 $this->catchException($e);
196 // If the table doesn't exist we should consider the item nonexistent.
197 return TRUE;
198 }
199 }
200
201 /**
202 * {@inheritdoc}
203 */
204 public function deleteItem($item) {
205 try {
206 $this->connection->delete(static::TABLE_NAME)
207 ->condition('item_id', $item->item_id)
208 ->execute();
209 }
210 catch (\Exception $e) {
211 $this->catchException($e);
212 }
213 }
214
215 /**
216 * {@inheritdoc}
217 */
218 public function createQueue() {
219 // All tasks are stored in a single database table (which is created on
220 // demand) so there is nothing we need to do to create a new queue.
221 }
222
223 /**
224 * {@inheritdoc}
225 */
226 public function deleteQueue() {
227 try {
228 $this->connection->delete(static::TABLE_NAME)
229 ->condition('name', $this->name)
230 ->execute();
231 }
232 catch (\Exception $e) {
233 $this->catchException($e);
234 }
235 }
236
237 /**
238 * {@inheritdoc}
239 */
240 public function garbageCollection() {
241 try {
242 // Clean up the queue for failed batches.
243 $this->connection->delete(static::TABLE_NAME)
244 ->condition('created', REQUEST_TIME - 864000, '<')
245 ->condition('name', 'drupal_batch:%', 'LIKE')
246 ->execute();
247
248 // Reset expired items in the default queue implementation table. If that's
249 // not used, this will simply be a no-op.
250 $this->connection->update(static::TABLE_NAME)
251 ->fields([
252 'expire' => 0,
253 ])
254 ->condition('expire', 0, '<>')
255 ->condition('expire', REQUEST_TIME, '<')
256 ->execute();
257 }
258 catch (\Exception $e) {
259 $this->catchException($e);
260 }
261 }
262
263 /**
264 * Check if the table exists and create it if not.
265 */
266 protected function ensureTableExists() {
267 try {
268 $database_schema = $this->connection->schema();
269 if (!$database_schema->tableExists(static::TABLE_NAME)) {
270 $schema_definition = $this->schemaDefinition();
271 $database_schema->createTable(static::TABLE_NAME, $schema_definition);
272 return TRUE;
273 }
274 }
275 // If another process has already created the queue table, attempting to
276 // recreate it will throw an exception. In this case just catch the
277 // exception and do nothing.
278 catch (DatabaseException $e) {
279 return TRUE;
280 }
281 return FALSE;
282 }
283
284 /**
285 * Act on an exception when queue might be stale.
286 *
287 * If the table does not yet exist, that's fine, but if the table exists and
288 * yet the query failed, then the queue is stale and the exception needs to
289 * propagate.
290 *
291 * @param $e
292 * The exception.
293 *
294 * @throws \Exception
295 * If the table exists the exception passed in is rethrown.
296 */
297 protected function catchException(\Exception $e) {
298 if ($this->connection->schema()->tableExists(static::TABLE_NAME)) {
299 throw $e;
300 }
301 }
302
303 /**
304 * Defines the schema for the queue table.
305 *
306 * @internal
307 */
308 public function schemaDefinition() {
309 return [
310 'description' => 'Stores items in queues.',
311 'fields' => [
312 'item_id' => [
313 'type' => 'serial',
314 'unsigned' => TRUE,
315 'not null' => TRUE,
316 'description' => 'Primary Key: Unique item ID.',
317 ],
318 'name' => [
319 'type' => 'varchar_ascii',
320 'length' => 255,
321 'not null' => TRUE,
322 'default' => '',
323 'description' => 'The queue name.',
324 ],
325 'data' => [
326 'type' => 'blob',
327 'not null' => FALSE,
328 'size' => 'big',
329 'serialize' => TRUE,
330 'description' => 'The arbitrary data for the item.',
331 ],
332 'expire' => [
333 'type' => 'int',
334 'not null' => TRUE,
335 'default' => 0,
336 'description' => 'Timestamp when the claim lease expires on the item.',
337 ],
338 'created' => [
339 'type' => 'int',
340 'not null' => TRUE,
341 'default' => 0,
342 'description' => 'Timestamp when the item was created.',
343 ],
344 ],
345 'primary key' => ['item_id'],
346 'indexes' => [
347 'name_created' => ['name', 'created'],
348 'expire' => ['expire'],
349 ],
350 ];
351 }
352
353}
354