CRM/Queue/Runner.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  * The queue runner is a helper which runs all jobs in a queue.
  14  *
  15  * The queue runner is most useful for one-off queues (such as an upgrade);
  16  * if the intention is to develop a dedicated, long-running worker thread,
  17  * then one should consider writing a new queue consumer.
  18  */
  19 class CRM_Queue_Runner {
  20
  21   /**
  22    * The failed task should be discarded, and queue processing should continue.
  23    */
  24   const ERROR_CONTINUE = 1;
  25
  26   /**
  27    * The failed task should be kept in the queue, and queue processing should
  28    * abort.
  29    */
  30   const ERROR_ABORT = 2;
  31
  32   /**
  33    * @var string
  34    */
  35   public $title;
  36
  37   /**
  38    * @var CRM_Queue_Queue
  39    */
  40   public $queue;
  41   public $errorMode;
  42   public $isMinimal;
  43   public $onEnd;
  44   public $onEndUrl;
  45   public $pathPrefix;
  46   /**
  47    * queue-runner id; used for persistence
  48    * @var int
  49    */
  50   public $qrid;
  51
  52   /**
  53    * @var array
  54    * Whether to display buttons, eg ('retry' => TRUE, 'skip' => FALSE)
  55    */
  56   public $buttons;
  57
  58   /**
  59    * @var CRM_Queue_TaskContext
  60    */
  61   public $taskCtx;
  62
  63   /**
  64    * Locate a previously-created instance of the queue-runner.
  65    *
  66    * @param string $qrid
  67    *   The queue-runner ID.
  68    *
  69    * @return CRM_Queue_Runner|NULL
  70    */
  71   public static function instance($qrid) {
  72     if (!empty($_SESSION['queueRunners'][$qrid])) {
  73       return unserialize($_SESSION['queueRunners'][$qrid]);
  74     }
  75     else {
  76       return NULL;
  77     }
  78   }
  79
  80   /**
  81    *
  82    * FIXME: parameter validation
  83    * FIXME: document signature of onEnd callback
  84    *
  85    * @param array $runnerSpec
  86    *   Array with keys:
  87    *   - queue: CRM_Queue_Queue
  88    *   - errorMode: int, ERROR_CONTINUE or ERROR_ABORT.
  89    *   - onEnd: mixed, a callback to update the UI after running; should be
  90    *     both callable and serializable.
  91    *   - onEndUrl: string, the URL to which one redirects.
  92    *   - pathPrefix: string, prepended to URLs for the web-runner;
  93    *     default: 'civicrm/queue'.
  94    */
  95   public function __construct($runnerSpec) {
  96     $this->title = CRM_Utils_Array::value('title', $runnerSpec, ts('Queue Runner'));
  97     $this->queue = $runnerSpec['queue'];
  98     $this->errorMode = CRM_Utils_Array::value('errorMode', $runnerSpec, self::ERROR_ABORT);
  99     $this->isMinimal = CRM_Utils_Array::value('isMinimal', $runnerSpec, FALSE);
 100     $this->onEnd = $runnerSpec['onEnd'] ?? NULL;
 101     $this->onEndUrl = $runnerSpec['onEndUrl'] ?? NULL;
 102     $this->pathPrefix = CRM_Utils_Array::value('pathPrefix', $runnerSpec, 'civicrm/queue');
 103     $this->buttons = CRM_Utils_Array::value('buttons', $runnerSpec, ['retry' => TRUE, 'skip' => TRUE]);
 104     // perhaps this value should be randomized?
 105     $this->qrid = $this->queue->getName();
 106   }
 107
 108   /**
 109    * @return array
 110    */
 111   public function __sleep() {
 112     // exclude taskCtx
 113     return [
 114       'title',
 115       'queue',
 116       'errorMode',
 117       'isMinimal',
 118       'onEnd',
 119       'onEndUrl',
 120       'pathPrefix',
 121       'qrid',
 122       'buttons',
 123     ];
 124   }
 125
 126   /**
 127    * Redirect to the web-based queue-runner and evaluate all tasks in a queue.
 128    */
 129   public function runAllViaWeb() {
 130     $_SESSION['queueRunners'][$this->qrid] = serialize($this);
 131     $url = CRM_Utils_System::url($this->pathPrefix . '/runner', 'reset=1&qrid=' . urlencode($this->qrid));
 132     CRM_Utils_System::redirect($url);
 133     // TODO: evaluate items incrementally via AJAX polling, cleanup session
 134   }
 135
 136   /**
 137    * Immediately run all tasks in a queue (until either reaching the end
 138    * of the queue or encountering an error)
 139    *
 140    * If the runner has an onEndUrl, then this function will not return
 141    *
 142    * @return mixed
 143    *   TRUE if all tasks complete normally; otherwise, an array describing the
 144    *   failed task
 145    */
 146   public function runAll() {
 147     $taskResult = $this->formatTaskResult(TRUE);
 148     while ($taskResult['is_continue']) {
 149       // setRaiseException should't be necessary here, but there's a bug
 150       // somewhere which causes this setting to be lost.  Observed while
 151       // upgrading 4.0=>4.2.  This preference really shouldn't be a global
 152       // setting -- it should be more of a contextual/stack-based setting.
 153       // This should be appropriate because queue-runners are not used with
 154       // basic web pages -- they're used with CLI/REST/AJAX.
 155       $errorScope = CRM_Core_TemporaryErrorScope::useException();
 156       $taskResult = $this->runNext();
 157       $errorScope = NULL;
 158     }
 159
 160     if ($taskResult['numberOfItems'] == 0) {
 161       $result = $this->handleEnd();
 162       if (!empty($result['redirect_url'])) {
 163         CRM_Utils_System::redirect($result['redirect_url']);
 164       }
 165       return TRUE;
 166     }
 167     else {
 168       return $taskResult;
 169     }
 170   }
 171
 172   /**
 173    * Take the next item from the queue and attempt to run it.
 174    *
 175    * Individual tasks may also throw exceptions -- caller should watch for
 176    * exceptions.
 177    *
 178    * @param bool $useSteal
 179    *   Whether to steal active locks.
 180    *
 181    * @return array
 182    *   - is_error => bool,
 183    *   - is_continue => bool,
 184    *   - numberOfItems => int,
 185    *   - 'last_task_title' => $,
 186    *   - 'exception' => $
 187    */
 188   public function runNext($useSteal = FALSE) {
 189     if ($useSteal) {
 190       $item = $this->queue->stealItem();
 191     }
 192     else {
 193       $item = $this->queue->claimItem();
 194     }
 195
 196     if ($item) {
 197       $this->lastTaskTitle = $item->data->title;
 198
 199       $exception = NULL;
 200       try {
 201         CRM_Core_Error::debug_log_message("Running task: " . $this->lastTaskTitle);
 202         $isOK = $item->data->run($this->getTaskContext());
 203         if (!$isOK) {
 204           $exception = new Exception('Task returned false');
 205         }
 206       }
 207       catch (Exception$e) {
 208         $isOK = FALSE;
 209         $exception = $e;
 210       }
 211
 212       if ($isOK) {
 213         $this->queue->deleteItem($item);
 214       }
 215       else {
 216         $this->releaseErrorItem($item);
 217       }
 218
 219       return $this->formatTaskResult($isOK, $exception);
 220     }
 221     else {
 222       return $this->formatTaskResult(FALSE, new Exception('Failed to claim next task'));
 223     }
 224   }
 225
 226   /**
 227    * Take the next item from the queue and attempt to run it.
 228    *
 229    * Individual tasks may also throw exceptions -- caller should watch for
 230    * exceptions.
 231    *
 232    * @param bool $useSteal
 233    *   Whether to steal active locks.
 234    *
 235    * @return array
 236    *   - is_error => bool,
 237    *   - is_continue => bool,
 238    *   - numberOfItems => int)
 239    */
 240   public function skipNext($useSteal = FALSE) {
 241     if ($useSteal) {
 242       $item = $this->queue->stealItem();
 243     }
 244     else {
 245       $item = $this->queue->claimItem();
 246     }
 247
 248     if ($item) {
 249       $this->lastTaskTitle = $item->data->title;
 250       $this->queue->deleteItem($item);
 251       return $this->formatTaskResult(TRUE);
 252     }
 253     else {
 254       return $this->formatTaskResult(FALSE, new Exception('Failed to claim next task'));
 255     }
 256   }
 257
 258   /**
 259    * Release an item in keeping with the error mode.
 260    *
 261    * @param object $item
 262    *   The item previously produced by Queue::claimItem.
 263    */
 264   protected function releaseErrorItem($item) {
 265     switch ($this->errorMode) {
 266       case self::ERROR_CONTINUE:
 267         $this->queue->deleteItem($item);
 268       case self::ERROR_ABORT:
 269       default:
 270         $this->queue->releaseItem($item);
 271     }
 272   }
 273
 274   /**
 275    * @return array
 276    *   - is_error => bool,
 277    *   - is_continue => bool,
 278    *   - numberOfItems => int,
 279    *   - redirect_url => string
 280    */
 281   public function handleEnd() {
 282     if (is_callable($this->onEnd)) {
 283       call_user_func($this->onEnd, $this->getTaskContext());
 284     }
 285     // Don't remove queueRunner until onEnd succeeds
 286     if (!empty($_SESSION['queueRunners'][$this->qrid])) {
 287       unset($_SESSION['queueRunners'][$this->qrid]);
 288     }
 289
 290     // Fallback; web UI does redirect in Javascript
 291     $result = [];
 292     $result['is_error'] = 0;
 293     $result['numberOfItems'] = 0;
 294     $result['is_continue'] = 0;
 295     if (!empty($this->onEndUrl)) {
 296       $result['redirect_url'] = $this->onEndUrl;
 297     }
 298     return $result;
 299   }
 300
 301   /**
 302    * Format a result record which describes whether the task completed.
 303    *
 304    * @param bool $isOK
 305    *   TRUE if the task completed successfully.
 306    * @param Exception|NULL $exception
 307    *   If applicable, an unhandled exception that arose during execution.
 308    *
 309    * @return array
 310    *   (is_error => bool, is_continue => bool, numberOfItems => int)
 311    */
 312   public function formatTaskResult($isOK, $exception = NULL) {
 313     $numberOfItems = $this->queue->numberOfItems();
 314
 315     $result = [];
 316     $result['is_error'] = $isOK ? 0 : 1;
 317     $result['exception'] = $exception;
 318     $result['last_task_title'] = $this->lastTaskTitle ?? '';
 319     $result['numberOfItems'] = $this->queue->numberOfItems();
 320     if ($result['numberOfItems'] <= 0) {
 321       // nothing to do
 322       $result['is_continue'] = 0;
 323     }
 324     elseif ($isOK) {
 325       // more tasks remain, and this task succeeded
 326       $result['is_continue'] = 1;
 327     }
 328     elseif ($this->errorMode == CRM_Queue_Runner::ERROR_CONTINUE) {
 329       // more tasks remain, and we can disregard this error
 330       $result['is_continue'] = 1;
 331     }
 332     else {
 333       // more tasks remain, but we can't disregard the error
 334       $result['is_continue'] = 0;
 335     }
 336
 337     return $result;
 338   }
 339
 340   /**
 341    * @return CRM_Queue_TaskContext
 342    */
 343   protected function getTaskContext() {
 344     if (!is_object($this->taskCtx)) {
 345       $this->taskCtx = new CRM_Queue_TaskContext();
 346       $this->taskCtx->queue = $this->queue;
 347       // $this->taskCtx->log = CRM_Core_Config::getLog();
 348       $this->taskCtx->log = CRM_Core_Error::createDebugLogger();
 349     }
 350     return $this->taskCtx;
 351   }
 352
 353 }