3 +--------------------------------------------------------------------+
4 | Copyright CiviCRM LLC. All rights reserved. |
6 | This work is published under the GNU AGPLv3 license with some |
7 | permitted exceptions and without any warranty. For full license |
8 | and copyright information, see https://civicrm.org/licensing |
9 +--------------------------------------------------------------------+
13 * A queue implementation which stores items in the CiviCRM SQL database
15 class CRM_Queue_Queue_SqlParallel
extends CRM_Queue_Queue
{
18 * Create a reference to queue. After constructing the queue, one should
19 * usually call createQueue (if it's a new queue) or loadQueue (if it's
20 * known to be an existing queue).
22 * @param array $queueSpec
24 * - type: string, required, e.g. "interactive", "immediate", "stomp",
26 * - name: string, required, e.g. "upgrade-tasks"
27 * - reset: bool, optional; if a queue is found, then it should be
28 * flushed; default to TRUE
29 * - (additional keys depending on the queue provider).
31 public function __construct($queueSpec) {
32 parent
::__construct($queueSpec);
36 * Perform any registation or resource-allocation for a new queue
38 public function createQueue() {
39 // nothing to do -- just start CRUDing items in the appropriate table
43 * Perform any loading or pre-fetch for an existing queue.
45 public function loadQueue() {
46 // nothing to do -- just start CRUDing items in the appropriate table
50 * Release any resources claimed by the queue (memory, DB rows, etc)
52 public function deleteQueue() {
53 return CRM_Core_DAO
::singleValueQuery("
54 DELETE FROM civicrm_queue_item
57 1 => [$this->getName(), 'String'],
62 * Check if the queue exists.
66 public function existsQueue() {
67 return ($this->numberOfItems() > 0);
71 * Add a new item to the queue.
74 * Serializable PHP object or array.
75 * @param array $options
76 * Queue-dependent options; for example, if this is a
77 * priority-queue, then $options might specify the item's priority.
79 public function createItem($data, $options = []) {
80 $dao = new CRM_Queue_DAO_QueueItem();
81 $dao->queue_name
= $this->getName();
82 $dao->submit_time
= CRM_Utils_Time
::getTime('YmdHis');
83 $dao->data
= serialize($data);
84 $dao->weight
= CRM_Utils_Array
::value('weight', $options, 0);
89 * Determine number of items remaining in the queue.
93 public function numberOfItems() {
94 return CRM_Core_DAO
::singleValueQuery("
96 FROM civicrm_queue_item
99 1 => [$this->getName(), 'String'],
106 * @param int $lease_time
110 * With key 'data' that matches the inputted data.
112 public function claimItem($lease_time = 3600) {
115 $dao = CRM_Core_DAO
::executeQuery('LOCK TABLES civicrm_queue_item WRITE;');
116 $sql = "SELECT id, queue_name, submit_time, release_time, data
117 FROM civicrm_queue_item
118 WHERE queue_name = %1
119 AND (release_time IS NULL OR release_time < %2)
120 ORDER BY weight ASC, id ASC
124 1 => [$this->getName(), 'String'],
125 2 => [CRM_Utils_Time
::getTime(), 'Timestamp'],
127 $dao = CRM_Core_DAO
::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
128 if (is_a($dao, 'DB_Error')) {
129 // FIXME - Adding code to allow tests to pass
130 CRM_Core_Error
::fatal();
134 $nowEpoch = CRM_Utils_Time
::getTimeRaw();
135 CRM_Core_DAO
::executeQuery("UPDATE civicrm_queue_item SET release_time = %1 WHERE id = %2", [
136 '1' => [date('YmdHis', $nowEpoch +
$lease_time), 'String'],
137 '2' => [$dao->id
, 'Integer'],
139 // (Comment by artfulrobot Sep 2019: Not sure what the below comment means, should be removed/clarified?)
140 // work-around: inconsistent date-formatting causes unintentional breakage
141 # $dao->submit_time = date('YmdHis', strtotime($dao->submit_time));
142 # $dao->release_time = date('YmdHis', $nowEpoch + $lease_time);
144 $dao->data
= unserialize($dao->data
);
148 $dao = CRM_Core_DAO
::executeQuery('UNLOCK TABLES;');
154 * Get the next item, even if there's an active lease
156 * @param int $lease_time
160 * With key 'data' that matches the inputted data.
162 public function stealItem($lease_time = 3600) {
164 SELECT id, queue_name, submit_time, release_time, data
165 FROM civicrm_queue_item
166 WHERE queue_name = %1
167 ORDER BY weight ASC, id ASC
171 1 => [$this->getName(), 'String'],
173 $dao = CRM_Core_DAO
::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
175 $nowEpoch = CRM_Utils_Time
::getTimeRaw();
176 CRM_Core_DAO
::executeQuery("UPDATE civicrm_queue_item SET release_time = %1 WHERE id = %2", [
177 '1' => [date('YmdHis', $nowEpoch +
$lease_time), 'String'],
178 '2' => [$dao->id
, 'Integer'],
180 $dao->data
= unserialize($dao->data
);
186 * Remove an item from the queue.
188 * @param CRM_Core_DAO $dao
189 * The item returned by claimItem.
191 public function deleteItem($dao) {
197 * Return an item that could not be processed.
199 * @param CRM_Core_DAO $dao
200 * The item returned by claimItem.
202 public function releaseItem($dao) {
203 $sql = "UPDATE civicrm_queue_item SET release_time = NULL WHERE id = %1";
205 1 => [$dao->id
, 'Integer'],
207 CRM_Core_DAO
::executeQuery($sql, $params);