CRM/Queue/Queue/Memory.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_Memory extends CRM_Queue_Queue {
  16
  17   /**
  18    * @var array
  19    *   array(queueItemId => queueItemData)
  20    */
  21   public $items;
  22
  23   /**
  24    * @var array
  25    *   array(queueItemId => releaseTime), expressed in seconds since epoch.
  26    */
  27   public $releaseTimes;
  28
  29   public $nextQueueItemId = 1;
  30
  31   /**
  32    * Create a reference to queue. After constructing the queue, one should
  33    * usually call createQueue (if it's a new queue) or loadQueue (if it's
  34    * known to be an existing queue).
  35    *
  36    * @param array $queueSpec
  37    *   Array with keys:
  38    *   - type: string, required, e.g. "interactive", "immediate", "stomp",
  39    *     "beanstalk"
  40    *   - name: string, required, e.g. "upgrade-tasks"
  41    *   - reset: bool, optional; if a queue is found, then it should be
  42    *     flushed; default to TRUE
  43    *   - (additional keys depending on the queue provider).
  44    */
  45   public function __construct($queueSpec) {
  46     parent::__construct($queueSpec);
  47   }
  48
  49   /**
  50    * Perform any registation or resource-allocation for a new queue
  51    */
  52   public function createQueue() {
  53     $this->items = [];
  54     $this->releaseTimes = [];
  55   }
  56
  57   /**
  58    * Perform any loading or pre-fetch for an existing queue.
  59    */
  60   public function loadQueue() {
  61     // $this->createQueue();
  62     throw new Exception('Unsupported: CRM_Queue_Queue_Memory::loadQueue');
  63   }
  64
  65   /**
  66    * Release any resources claimed by the queue (memory, DB rows, etc)
  67    */
  68   public function deleteQueue() {
  69     $this->items = NULL;
  70     $this->releaseTimes = NULL;
  71   }
  72
  73   /**
  74    * Check if the queue exists.
  75    *
  76    * @return bool
  77    */
  78   public function existsQueue() {
  79     return is_array($this->items);
  80   }
  81
  82   /**
  83    * Add a new item to the queue.
  84    *
  85    * @param mixed $data
  86    *   Serializable PHP object or array.
  87    * @param array $options
  88    *   Queue-dependent options; for example, if this is a
  89    *   priority-queue, then $options might specify the item's priority.
  90    */
  91   public function createItem($data, $options = []) {
  92     $id = $this->nextQueueItemId++;
  93     // force copy, no unintendedsharing effects from pointers
  94     $this->items[$id] = serialize($data);
  95   }
  96
  97   /**
  98    * Determine number of items remaining in the queue.
  99    *
 100    * @return int
 101    */
 102   public function numberOfItems() {
 103     return count($this->items);
 104   }
 105
 106   /**
 107    * Get and remove the next item.
 108    *
 109    * @param int $leaseTime
 110    *   Seconds.
 111    *
 112    * @return object
 113    *   Includes key 'data' that matches the inputted data.
 114    */
 115   public function claimItem($leaseTime = 3600) {
 116     // foreach hits the items in order -- but we short-circuit after the first
 117     foreach ($this->items as $id => $data) {
 118       $nowEpoch = CRM_Utils_Time::getTimeRaw();
 119       if (empty($this->releaseTimes[$id]) || $this->releaseTimes[$id] < $nowEpoch) {
 120         $this->releaseTimes[$id] = $nowEpoch + $leaseTime;
 121
 122         $item = new stdClass();
 123         $item->id = $id;
 124         $item->data = unserialize($data);
 125         return $item;
 126       }
 127       else {
 128         // item in queue is reserved
 129         return FALSE;
 130       }
 131     }
 132     // nothing in queue
 133     return FALSE;
 134   }
 135
 136   /**
 137    * Get the next item.
 138    *
 139    * @param int $leaseTime
 140    *   Seconds.
 141    *
 142    * @return object
 143    *   With key 'data' that matches the inputted data.
 144    */
 145   public function stealItem($leaseTime = 3600) {
 146     // foreach hits the items in order -- but we short-circuit after the first
 147     foreach ($this->items as $id => $data) {
 148       $nowEpoch = CRM_Utils_Time::getTimeRaw();
 149       $this->releaseTimes[$id] = $nowEpoch + $leaseTime;
 150
 151       $item = new stdClass();
 152       $item->id = $id;
 153       $item->data = unserialize($data);
 154       return $item;
 155     }
 156     // nothing in queue
 157     return FALSE;
 158   }
 159
 160   /**
 161    * Remove an item from the queue.
 162    *
 163    * @param object $item
 164    *   The item returned by claimItem.
 165    */
 166   public function deleteItem($item) {
 167     unset($this->items[$item->id]);
 168     unset($this->releaseTimes[$item->id]);
 169   }
 170
 171   /**
 172    * Return an item that could not be processed.
 173    *
 174    * @param CRM_Core_DAO $item
 175    *   The item returned by claimItem.
 176    */
 177   public function releaseItem($item) {
 178     unset($this->releaseTimes[$item->id]);
 179   }
 180
 181 }