CRM/Queue/Queue/Sql.php

   1 <?php
   2 /*
   3  +--------------------------------------------------------------------+
   4  | Copyright CiviCRM LLC. All rights reserved.                        |
   5  |                                                                    |
   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  +--------------------------------------------------------------------+
  10  */
  11
  12 /**
  13  * A queue implementation which stores items in the CiviCRM SQL database
  14  */
  15 class CRM_Queue_Queue_Sql extends CRM_Queue_Queue {
  16
  17   /**
  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).
  21    *
  22    * @param array $queueSpec
  23    *   Array with keys:
  24    *   - type: string, required, e.g. "interactive", "immediate", "stomp",
  25    *     "beanstalk"
  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).
  30    */
  31   public function __construct($queueSpec) {
  32     parent::__construct($queueSpec);
  33   }
  34
  35   /**
  36    * Perform any registation or resource-allocation for a new queue
  37    */
  38   public function createQueue() {
  39     // nothing to do -- just start CRUDing items in the appropriate table
  40   }
  41
  42   /**
  43    * Perform any loading or pre-fetch for an existing queue.
  44    */
  45   public function loadQueue() {
  46     // nothing to do -- just start CRUDing items in the appropriate table
  47   }
  48
  49   /**
  50    * Release any resources claimed by the queue (memory, DB rows, etc)
  51    */
  52   public function deleteQueue() {
  53     return CRM_Core_DAO::singleValueQuery("
  54       DELETE FROM civicrm_queue_item
  55       WHERE queue_name = %1
  56     ", [
  57       1 => [$this->getName(), 'String'],
  58     ]);
  59   }
  60
  61   /**
  62    * Check if the queue exists.
  63    *
  64    * @return bool
  65    */
  66   public function existsQueue() {
  67     return ($this->numberOfItems() > 0);
  68   }
  69
  70   /**
  71    * Add a new item to the queue.
  72    *
  73    * @param mixed $data
  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.
  78    */
  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);
  85     $dao->save();
  86   }
  87
  88   /**
  89    * Determine number of items remaining in the queue.
  90    *
  91    * @return int
  92    */
  93   public function numberOfItems() {
  94     return CRM_Core_DAO::singleValueQuery("
  95       SELECT count(*)
  96       FROM civicrm_queue_item
  97       WHERE queue_name = %1
  98     ", [
  99       1 => [$this->getName(), 'String'],
 100     ]);
 101   }
 102
 103   /**
 104    * Get the next item.
 105    *
 106    * @param int $lease_time
 107    *   Seconds.
 108    *
 109    * @return object
 110    *   With key 'data' that matches the inputted data.
 111    */
 112   public function claimItem($lease_time = 3600) {
 113
 114     $result = NULL;
 115     $dao = CRM_Core_DAO::executeQuery('LOCK TABLES civicrm_queue_item WRITE;');
 116     $sql = "
 117         SELECT first_in_queue.* FROM (
 118           SELECT id, queue_name, submit_time, release_time, data
 119           FROM civicrm_queue_item
 120           WHERE queue_name = %1
 121           ORDER BY weight ASC, id ASC
 122           LIMIT 1
 123         ) first_in_queue
 124         WHERE release_time IS NULL OR release_time < %2
 125       ";
 126     $params = [
 127       1 => [$this->getName(), 'String'],
 128       2 => [CRM_Utils_Time::getTime(), 'Timestamp'],
 129     ];
 130     $dao = CRM_Core_DAO::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
 131     if (is_a($dao, 'DB_Error')) {
 132       // FIXME - Adding code to allow tests to pass
 133       throw new CRM_Core_Exception('Unable to claim queue item');
 134     }
 135
 136     if ($dao->fetch()) {
 137       $nowEpoch = CRM_Utils_Time::getTimeRaw();
 138       CRM_Core_DAO::executeQuery("UPDATE civicrm_queue_item SET release_time = %1 WHERE id = %2", [
 139         '1' => [date('YmdHis', $nowEpoch + $lease_time), 'String'],
 140         '2' => [$dao->id, 'Integer'],
 141       ]);
 142       // (Comment by artfulrobot Sep 2019: Not sure what the below comment means, should be removed/clarified?)
 143       // work-around: inconsistent date-formatting causes unintentional breakage
 144       #        $dao->submit_time = date('YmdHis', strtotime($dao->submit_time));
 145       #        $dao->release_time = date('YmdHis', $nowEpoch + $lease_time);
 146       #        $dao->save();
 147       $dao->data = unserialize($dao->data);
 148       $result = $dao;
 149     }
 150
 151     $dao = CRM_Core_DAO::executeQuery('UNLOCK TABLES;');
 152
 153     return $result;
 154   }
 155
 156   /**
 157    * Get the next item, even if there's an active lease
 158    *
 159    * @param int $lease_time
 160    *   Seconds.
 161    *
 162    * @return object
 163    *   With key 'data' that matches the inputted data.
 164    */
 165   public function stealItem($lease_time = 3600) {
 166     $sql = "
 167       SELECT id, queue_name, submit_time, release_time, data
 168       FROM civicrm_queue_item
 169       WHERE queue_name = %1
 170       ORDER BY weight ASC, id ASC
 171       LIMIT 1
 172     ";
 173     $params = [
 174       1 => [$this->getName(), 'String'],
 175     ];
 176     $dao = CRM_Core_DAO::executeQuery($sql, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
 177     if ($dao->fetch()) {
 178       $nowEpoch = CRM_Utils_Time::getTimeRaw();
 179       CRM_Core_DAO::executeQuery("UPDATE civicrm_queue_item SET release_time = %1 WHERE id = %2", [
 180         '1' => [date('YmdHis', $nowEpoch + $lease_time), 'String'],
 181         '2' => [$dao->id, 'Integer'],
 182       ]);
 183       $dao->data = unserialize($dao->data);
 184       return $dao;
 185     }
 186   }
 187
 188   /**
 189    * Remove an item from the queue.
 190    *
 191    * @param CRM_Core_DAO $dao
 192    *   The item returned by claimItem.
 193    */
 194   public function deleteItem($dao) {
 195     $dao->delete();
 196   }
 197
 198   /**
 199    * Return an item that could not be processed.
 200    *
 201    * @param CRM_Core_DAO $dao
 202    *   The item returned by claimItem.
 203    */
 204   public function releaseItem($dao) {
 205     $sql = "UPDATE civicrm_queue_item SET release_time = NULL WHERE id = %1";
 206     $params = [
 207       1 => [$dao->id, 'Integer'],
 208     ];
 209     CRM_Core_DAO::executeQuery($sql, $params);
 210   }
 211
 212 }