CRM/Queue/Queue/Sql.php

   1 <?php
   2 /*
   3  +--------------------------------------------------------------------+
   4  | CiviCRM version 5                                                  |
   5  +--------------------------------------------------------------------+
   6  | Copyright CiviCRM LLC (c) 2004-2019                                |
   7  +--------------------------------------------------------------------+
   8  | This file is a part of CiviCRM.                                    |
   9  |                                                                    |
  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.   |
  13  |                                                                    |
  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.        |
  18  |                                                                    |
  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  +--------------------------------------------------------------------+
  26  */
  27
  28 /**
  29  * A queue implementation which stores items in the CiviCRM SQL database
  30  */
  31 class CRM_Queue_Queue_Sql extends CRM_Queue_Queue {
  32
  33   /**
  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).
  37    *
  38    * @param array $queueSpec
  39    *   Array with keys:
  40    *   - type: string, required, e.g. "interactive", "immediate", "stomp",
  41    *     "beanstalk"
  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).
  46    */
  47   public function __construct($queueSpec) {
  48     parent::__construct($queueSpec);
  49   }
  50
  51   /**
  52    * Perform any registation or resource-allocation for a new queue
  53    */
  54   public function createQueue() {
  55     // nothing to do -- just start CRUDing items in the appropriate table
  56   }
  57
  58   /**
  59    * Perform any loading or pre-fetch for an existing queue.
  60    */
  61   public function loadQueue() {
  62     // nothing to do -- just start CRUDing items in the appropriate table
  63   }
  64
  65   /**
  66    * Release any resources claimed by the queue (memory, DB rows, etc)
  67    */
  68   public function deleteQueue() {
  69     return CRM_Core_DAO::singleValueQuery("
  70       DELETE FROM civicrm_queue_item
  71       WHERE queue_name = %1
  72     ", [
  73       1 => [$this->getName(), 'String'],
  74     ]);
  75   }
  76
  77   /**
  78    * Check if the queue exists.
  79    *
  80    * @return bool
  81    */
  82   public function existsQueue() {
  83     return ($this->numberOfItems() > 0);
  84   }
  85
  86   /**
  87    * Add a new item to the queue.
  88    *
  89    * @param mixed $data
  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.
  94    */
  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);
 101     $dao->save();
 102   }
 103
 104   /**
 105    * Determine number of items remaining in the queue.
 106    *
 107    * @return int
 108    */
 109   public function numberOfItems() {
 110     return CRM_Core_DAO::singleValueQuery("
 111       SELECT count(*)
 112       FROM civicrm_queue_item
 113       WHERE queue_name = %1
 114     ", [
 115       1 => [$this->getName(), 'String'],
 116     ]);
 117   }
 118
 119   /**
 120    * Get the next item.
 121    *
 122    * @param int $lease_time
 123    *   Seconds.
 124    *
 125    * @return object
 126    *   With key 'data' that matches the inputted data.
 127    */
 128   public function claimItem($lease_time = 3600) {
 129
 130     $result = NULL;
 131     $dao = CRM_Core_DAO::executeQuery('LOCK TABLES civicrm_queue_item WRITE;');
 132     $sql = "
 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
 138           LIMIT 1
 139         ) first_in_queue
 140         WHERE release_time IS NULL OR release_time < %2
 141       ";
 142     $params = [
 143       1 => [$this->getName(), 'String'],
 144       2 => [CRM_Utils_Time::getTime(), 'Timestamp'],
 145     ];
 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();
 150     }
 151
 152     if ($dao->fetch()) {
 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'],
 157       ]);
 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);
 162       #        $dao->save();
 163       $dao->data = CRM_Utils_String::unserialize($dao->data);
 164       $result = $dao;
 165     }
 166
 167     $dao = CRM_Core_DAO::executeQuery('UNLOCK TABLES;');
 168
 169     return $result;
 170   }
 171
 172   /**
 173    * Get the next item, even if there's an active lease
 174    *
 175    * @param int $lease_time
 176    *   Seconds.
 177    *
 178    * @return object
 179    *   With key 'data' that matches the inputted data.
 180    */
 181   public function stealItem($lease_time = 3600) {
 182     $sql = "
 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
 187       LIMIT 1
 188     ";
 189     $params = [
 190       1 => [$this->getName(), 'String'],
 191     ];
 192     $dao = CRM_Core_DAO::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
 193     if ($dao->fetch()) {
 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'],
 198       ]);
 199       $dao->data = CRM_Utils_String::unserialize($dao->data);
 200       return $dao;
 201     }
 202   }
 203
 204   /**
 205    * Remove an item from the queue.
 206    *
 207    * @param CRM_Core_DAO $dao
 208    *   The item returned by claimItem.
 209    */
 210   public function deleteItem($dao) {
 211     $dao->delete();
 212   }
 213
 214   /**
 215    * Return an item that could not be processed.
 216    *
 217    * @param CRM_Core_DAO $dao
 218    *   The item returned by claimItem.
 219    */
 220   public function releaseItem($dao) {
 221     $sql = "UPDATE civicrm_queue_item SET release_time = NULL WHERE id = %1";
 222     $params = [
 223       1 => [$dao->id, 'Integer'],
 224     ];
 225     CRM_Core_DAO::executeQuery($sql, $params);
 226   }
 227
 228 }