function DatabaseQueue::claimItem

Overrides QueueInterface::claimItem

File

core/lib/Drupal/Core/Queue/DatabaseQueue.php, line 117

Class

DatabaseQueue

Default queue implementation.

Namespace

Code

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) {
        try {
            $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();
        } catch (\Exception $e) {
            $this->catchException($e);
        }
        // If the table does not exist there are no items currently available to
        // claim.
        if (empty($item)) {
            return FALSE;
        }
        // Try to update the item. Only one thread can succeed in UPDATEing the
        // same row. We cannot rely on \Drupal::time()->getRequestTime() because
        // items might be claimed by a single consumer which runs longer than 1
        // second. If we continue to use ::getRequestTime() instead of
        // ::getCurrentTime(), we steal time from the lease, and will tend to
        // reset items before the lease should really expire.
        $update = $this->connection
            ->update(static::TABLE_NAME)
            ->fields([
            'expire' => \Drupal::time()->getCurrentTime() + $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;
        }
    }
}