CRM/Queue/Queue/Memory.php

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