3 +--------------------------------------------------------------------+
5 +--------------------------------------------------------------------+
6 | Copyright CiviCRM LLC (c) 2004-2019 |
7 +--------------------------------------------------------------------+
8 | This file is a part of CiviCRM. |
10 | CiviCRM is free software; you can copy, modify, and distribute it |
11 | under the terms of the GNU Affero General Public License |
12 | Version 3, 19 November 2007 and the CiviCRM Licensing Exception. |
14 | CiviCRM is distributed in the hope that it will be useful, but |
15 | WITHOUT ANY WARRANTY; without even the implied warranty of |
16 | MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. |
17 | See the GNU Affero General Public License for more details. |
19 | You should have received a copy of the GNU Affero General Public |
20 | License and the CiviCRM Licensing Exception along |
21 | with this program; if not, contact CiviCRM LLC |
22 | at info[AT]civicrm[DOT]org. If you have questions about the |
23 | GNU Affero General Public License or the licensing of CiviCRM, |
24 | see the CiviCRM license FAQ at http://civicrm.org/licensing |
25 +--------------------------------------------------------------------+
29 * A queue implementation which stores items in the CiviCRM SQL database
31 class CRM_Queue_Queue_Sql
extends CRM_Queue_Queue
{
34 * Create a reference to queue. After constructing the queue, one should
35 * usually call createQueue (if it's a new queue) or loadQueue (if it's
36 * known to be an existing queue).
38 * @param array $queueSpec
40 * - type: string, required, e.g. "interactive", "immediate", "stomp",
42 * - name: string, required, e.g. "upgrade-tasks"
43 * - reset: bool, optional; if a queue is found, then it should be
44 * flushed; default to TRUE
45 * - (additional keys depending on the queue provider).
47 public function __construct($queueSpec) {
48 parent
::__construct($queueSpec);
52 * Perform any registation or resource-allocation for a new queue
54 public function createQueue() {
55 // nothing to do -- just start CRUDing items in the appropriate table
59 * Perform any loading or pre-fetch for an existing queue.
61 public function loadQueue() {
62 // nothing to do -- just start CRUDing items in the appropriate table
66 * Release any resources claimed by the queue (memory, DB rows, etc)
68 public function deleteQueue() {
69 return CRM_Core_DAO
::singleValueQuery("
70 DELETE FROM civicrm_queue_item
73 1 => [$this->getName(), 'String'],
78 * Check if the queue exists.
82 public function existsQueue() {
83 return ($this->numberOfItems() > 0);
87 * Add a new item to the queue.
90 * Serializable PHP object or array.
91 * @param array $options
92 * Queue-dependent options; for example, if this is a
93 * priority-queue, then $options might specify the item's priority.
95 public function createItem($data, $options = []) {
96 $dao = new CRM_Queue_DAO_QueueItem();
97 $dao->queue_name
= $this->getName();
98 $dao->submit_time
= CRM_Utils_Time
::getTime('YmdHis');
99 $dao->data
= serialize($data);
100 $dao->weight
= CRM_Utils_Array
::value('weight', $options, 0);
105 * Determine number of items remaining in the queue.
109 public function numberOfItems() {
110 return CRM_Core_DAO
::singleValueQuery("
112 FROM civicrm_queue_item
113 WHERE queue_name = %1
115 1 => [$this->getName(), 'String'],
122 * @param int $lease_time
126 * With key 'data' that matches the inputted data.
128 public function claimItem($lease_time = 3600) {
131 $dao = CRM_Core_DAO
::executeQuery('LOCK TABLES civicrm_queue_item WRITE;');
133 SELECT first_in_queue.* FROM (
134 SELECT id, queue_name, submit_time, release_time, data
135 FROM civicrm_queue_item
136 WHERE queue_name = %1
137 ORDER BY weight ASC, id ASC
140 WHERE release_time IS NULL OR release_time < %2
143 1 => [$this->getName(), 'String'],
144 2 => [CRM_Utils_Time
::getTime(), 'Timestamp'],
146 $dao = CRM_Core_DAO
::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
147 if (is_a($dao, 'DB_Error')) {
148 // FIXME - Adding code to allow tests to pass
149 CRM_Core_Error
::fatal();
153 $nowEpoch = CRM_Utils_Time
::getTimeRaw();
154 CRM_Core_DAO
::executeQuery("UPDATE civicrm_queue_item SET release_time = %1 WHERE id = %2", [
155 '1' => [date('YmdHis', $nowEpoch +
$lease_time), 'String'],
156 '2' => [$dao->id
, 'Integer'],
158 // (Comment by artfulrobot Sep 2019: Not sure what the below comment means, should be removed/clarified?)
159 // work-around: inconsistent date-formatting causes unintentional breakage
160 # $dao->submit_time = date('YmdHis', strtotime($dao->submit_time));
161 # $dao->release_time = date('YmdHis', $nowEpoch + $lease_time);
163 $dao->data
= unserialize($dao->data
);
167 $dao = CRM_Core_DAO
::executeQuery('UNLOCK TABLES;');
173 * Get the next item, even if there's an active lease
175 * @param int $lease_time
179 * With key 'data' that matches the inputted data.
181 public function stealItem($lease_time = 3600) {
183 SELECT id, queue_name, submit_time, release_time, data
184 FROM civicrm_queue_item
185 WHERE queue_name = %1
186 ORDER BY weight ASC, id ASC
190 1 => [$this->getName(), 'String'],
192 $dao = CRM_Core_DAO
::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
194 $nowEpoch = CRM_Utils_Time
::getTimeRaw();
195 CRM_Core_DAO
::executeQuery("UPDATE civicrm_queue_item SET release_time = %1 WHERE id = %2", [
196 '1' => [date('YmdHis', $nowEpoch +
$lease_time), 'String'],
197 '2' => [$dao->id
, 'Integer'],
199 $dao->data
= unserialize($dao->data
);
205 * Remove an item from the queue.
207 * @param CRM_Core_DAO $dao
208 * The item returned by claimItem.
210 public function deleteItem($dao) {
215 * Return an item that could not be processed.
217 * @param CRM_Core_DAO $dao
218 * The item returned by claimItem.
220 public function releaseItem($dao) {
221 $sql = "UPDATE civicrm_queue_item SET release_time = NULL WHERE id = %1";
223 1 => [$dao->id
, 'Integer'],
225 CRM_Core_DAO
::executeQuery($sql, $params);