[Summary view] [Print] [Text view]

   1  <?php /* $Id: event_queue.class.php 137 2008-04-04 16:12:02Z pedroix $ $URL: https://web2project.svn.sourceforge.net/svnroot/web2project/trunk/classes/event_queue.class.php $ */
   2  
   3  /**

   4   * Event handling queue class.

   5   *

   6   * The event queue uses the table event_queue to manage

   7   * event notifications and other timed events, as well as

   8   * outgoing emails.

   9   *

  10   * Copyright 2005, the web2Project team.

  11   */
  12  
  13  if (!defined('W2P_BASE_DIR')) {
  14      die('You should not access this file directly.');
  15  }
  16  
  17  class EventQueue {
  18  
  19      var $table = 'event_queue';
  20      var $update_list = array();
  21      var $delete_list = array();
  22      var $event_count = 0;
  23  
  24  	function EventQueue() {
  25      }
  26  
  27      /**

  28       * Add an event to the queue.

  29       *

  30       * The callback can either be the name of a global function or the

  31       * name of a class

  32       * @param mixed $callback function to call when this event is due.

  33       * @param mixed $args Arguments to pass to the callback

  34       * @param string $module module, or originator of the event

  35       * @param string $type type of event (to allow searching)

  36       * @param integer $id id of originating event.

  37       * @param integer $date Seconds since 1970 to trigger event.

  38       * @param integer $repeat_interval seconds to repeat

  39       * @param integer $repeat_count number of times to repeat

  40       * @return integer queue id

  41       */
  42  	function add($callback, &$args, $module, $sysmodule = false, $id = 0, $type = '', $date = 0, $repeat_interval = 0, $repeat_count = 1) {
  43          global $AppUI;
  44  
  45          if (!isset($AppUI)) {
  46              $user_id = 0;
  47          } else {
  48              $user_id = $AppUI->user_id;
  49          }
  50  
  51          if (is_array($callback)) {
  52              list($class, $method) = $callback;
  53              if (is_object($class)) {
  54                  $class = get_class($class);
  55              }
  56              $caller = $class . '::' . $method;
  57          } else {
  58              $caller = $callback;
  59          }
  60  
  61          $q = new DBQuery;
  62          $q->addTable($this->table);
  63          $q->addInsert('queue_owner', $user_id);
  64          $q->addInsert('queue_start', $date);
  65          $q->addInsert('queue_callback', $caller);
  66          $q->addInsert('queue_data', serialize($args));
  67          $q->addInsert('queue_repeat_interval', $repeat_interval);
  68          $q->addInsert('queue_repeat_count', $repeat_count);
  69          $q->addInsert('queue_module', $module);
  70          $q->addInsert('queue_type', $type);
  71          $q->addInsert('queue_origin_id', $id);
  72          if ($sysmodule) {
  73              $q->addInsert('queue_module_type', 'system');
  74          } else {
  75              $q->addInsert('queue_module_type', 'module');
  76          }
  77          if ($q->exec()) {
  78              $return = db_insert_id();
  79          } else {
  80              $return = false;
  81          }
  82          $q->clear();
  83          return $return;
  84      }
  85  
  86      /**

  87       * Remove the event from the queue. 

  88       * 

  89       */
  90  	function remove($id) {
  91          $q = new DBQuery;
  92          $q->setDelete($this->table);
  93          $q->addWhere('queue_id = \'' . $id . '\'');
  94          $q->exec();
  95          $q->clear();
  96      }
  97  
  98      /**

  99       * Find a queue record (or records) based upon the

 100       * 

 101       */
 102  	function find($module, $type, $id = null) {
 103          $q = new DBQuery;
 104          $q->addTable($this->table);
 105          $q->addWhere('queue_module = \'' . $module . '\'');
 106          $q->addWhere('queue_type = \'' . $type . '\'');
 107          if (isset($id)) {
 108              $q->addWhere('queue_origin_id = \'' . $id . '\'');
 109          }
 110          return $q->loadHashList('queue_id');
 111      }
 112  
 113      /**

 114       * Execute a queue entry.  This involves resolving the

 115       * method to execute and passing the arguments to it.

 116       */
 117  	function execute(&$fields) {
 118          global $AppUI;
 119  
 120          if (isset($fields['queue_module_type']) && $fields['queue_module_type'] == 'system') {
 121              include_once $AppUI->getSystemClass($fields['queue_module']);
 122          } else {
 123              include_once $AppUI->getModuleClass($fields['queue_module']);
 124          }
 125  
 126          $args = unserialize($fields['queue_data']);
 127          if (strpos($fields['queue_callback'], '::') !== false) {
 128              list($class, $method) = explode('::', $fields['queue_callback']);
 129              if (!class_exists($class)) {
 130                  dprint(__file__, __line__, 2, 'Cannot process event: Class ' . $class . ' does not exist');
 131                  return false;
 132              }
 133              $object = new $class;
 134              if (!method_exists($object, $method)) {
 135                  dprint(__file__, __line__, 2, 'Cannot process event: Method ' . $class . '::' . $method . ' does not exist');
 136                  return false;
 137              }
 138              return $object->$method($fields['queue_module'], $fields['queue_type'], $fields['queue_origin_id'], $fields['queue_owner'], $args);
 139          } else {
 140              $method = $fields['queue_callback'];
 141              if (!function_exists($method)) {
 142                  dprint(__file__, __line__, 2, 'Cannot process event: Function ' . $method . ' does not exist');
 143                  return false;
 144              }
 145              return $method($fields['queue_module'], $fields['queue_type'], $fields['queue_origin_id'], $fields['queue_owner'], $args);
 146          }
 147      }
 148  
 149      /**

 150       * Scans the queue for entries that are older than current date.

 151       * If it finds one it tries to execute the attached function.

 152       * If successful, the entry is removed from the queue, or if

 153       * it is a repeatable event the repeat time is added to the

 154       * start time and the repeat count (if set) is decremented.

 155       */
 156  	function scan() {
 157          $q = new DBQuery;
 158          $q->addTable($this->table);
 159          $now = time();
 160          $q->addWhere('queue_start < ' . $now);
 161          $rid = $q->exec();
 162  
 163          $this->event_count = 0;
 164          for ($rid; !$rid->EOF; $rid->moveNext()) {
 165              if ($this->execute($rid->fields)) {
 166                  $this->update_event($rid->fields);
 167                  $this->event_count++;
 168              }
 169          }
 170          $q->clear();
 171  
 172          $this->commit_updates();
 173      }
 174  
 175  	function update_event(&$fields) {
 176          if ($fields['queue_repeat_interval'] > 0 && $fields['queue_repeat_count'] > 0) {
 177              $fields['queue_start'] += $fields['queue_repeat_interval'];
 178              $fields['queue_repeat_count']--;
 179              $this->update_list[] = $fields;
 180          } else {
 181              $this->delete_list[] = $fields['queue_id'];
 182          }
 183      }
 184  
 185  	function commit_updates() {
 186          $q = new DBQuery;
 187          if (count($this->delete_list)) {
 188              $q->setDelete($this->table);
 189              $q->addWhere('queue_id IN (' . implode(',', $this->delete_list) . ')');
 190              $q->exec();
 191              $q->clear();
 192          }
 193          $this->delete_list = array();
 194  
 195          foreach ($this->update_list as $fields) {
 196              $q->addTable($this->table);
 197              $q->addUpdate('queue_repeat_count', $fields['queue_repeat_count']);
 198              $q->addUpdate('queue_start', $fields['queue_start']);
 199              $q->addWhere('queue_id = ' . $fields['queue_id']);
 200              $q->exec();
 201              $q->clear();
 202          }
 203          $this->update_list = array();
 204      }
 205  
 206  }
 207  ?>