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       $taskResult = $this->runNext();
 156       $errorScope = NULL;
 157     }
 158
 159     if ($taskResult['numberOfItems'] == 0) {
 160       $result = $this->handleEnd();
 161       if (!empty($result['redirect_url'])) {
 162         CRM_Utils_System::redirect($result['redirect_url']);
 163       }
 164       return TRUE;
 165     }
 166     else {
 167       return $taskResult;
 168     }
 169   }
 170
 171   /**
 172    * Take the next item from the queue and attempt to run it.
 173    *
 174    * Individual tasks may also throw exceptions -- caller should watch for
 175    * exceptions.
 176    *
 177    * @param bool $useSteal
 178    *   Whether to steal active locks.
 179    *
 180    * @return array
 181    *   - is_error => bool,
 182    *   - is_continue => bool,
 183    *   - numberOfItems => int,
 184    *   - 'last_task_title' => $,
 185    *   - 'exception' => $
 186    */
 187   public function runNext($useSteal = FALSE) {
 188     if ($useSteal) {
 189       $item = $this->queue->stealItem();
 190     }
 191     else {
 192       $item = $this->queue->claimItem();
 193     }
 194
 195     if ($item) {
 196       $this->lastTaskTitle = $item->data->title;
 197
 198       $exception = NULL;
 199       try {
 200         CRM_Core_Error::debug_log_message("Running task: " . $this->lastTaskTitle);
 201         $isOK = $item->data->run($this->getTaskContext());
 202         if (!$isOK) {
 203           $exception = new Exception('Task returned false');
 204         }
 205       }
 206       catch (Exception$e) {
 207         $isOK = FALSE;
 208         $exception = $e;
 209       }
 210
 211       if ($isOK) {
 212         $this->queue->deleteItem($item);
 213       }
 214       else {
 215         $this->releaseErrorItem($item);
 216       }
 217
 218       return $this->formatTaskResult($isOK, $exception);
 219     }
 220     else {
 221       return $this->formatTaskResult(FALSE, new Exception('Failed to claim next task'));
 222     }
 223   }
 224
 225   /**
 226    * Take the next item from the queue and attempt to run it.
 227    *
 228    * Individual tasks may also throw exceptions -- caller should watch for
 229    * exceptions.
 230    *
 231    * @param bool $useSteal
 232    *   Whether to steal active locks.
 233    *
 234    * @return array
 235    *   - is_error => bool,
 236    *   - is_continue => bool,
 237    *   - numberOfItems => int)
 238    */
 239   public function skipNext($useSteal = FALSE) {
 240     if ($useSteal) {
 241       $item = $this->queue->stealItem();
 242     }
 243     else {
 244       $item = $this->queue->claimItem();
 245     }
 246
 247     if ($item) {
 248       $this->lastTaskTitle = $item->data->title;
 249       $this->queue->deleteItem($item);
 250       return $this->formatTaskResult(TRUE);
 251     }
 252     else {
 253       return $this->formatTaskResult(FALSE, new Exception('Failed to claim next task'));
 254     }
 255   }
 256
 257   /**
 258    * Release an item in keeping with the error mode.
 259    *
 260    * @param object $item
 261    *   The item previously produced by Queue::claimItem.
 262    */
 263   protected function releaseErrorItem($item) {
 264     switch ($this->errorMode) {
 265       case self::ERROR_CONTINUE:
 266         $this->queue->deleteItem($item);
 267       case self::ERROR_ABORT:
 268       default:
 269         $this->queue->releaseItem($item);
 270     }
 271   }
 272
 273   /**
 274    * @return array
 275    *   - is_error => bool,
 276    *   - is_continue => bool,
 277    *   - numberOfItems => int,
 278    *   - redirect_url => string
 279    */
 280   public function handleEnd() {
 281     if (is_callable($this->onEnd)) {
 282       call_user_func($this->onEnd, $this->getTaskContext());
 283     }
 284     // Don't remove queueRunner until onEnd succeeds
 285     if (!empty($_SESSION['queueRunners'][$this->qrid])) {
 286       unset($_SESSION['queueRunners'][$this->qrid]);
 287     }
 288
 289     // Fallback; web UI does redirect in Javascript
 290     $result = [];
 291     $result['is_error'] = 0;
 292     $result['numberOfItems'] = 0;
 293     $result['is_continue'] = 0;
 294     if (!empty($this->onEndUrl)) {
 295       $result['redirect_url'] = $this->onEndUrl;
 296     }
 297     return $result;
 298   }
 299
 300   /**
 301    * Format a result record which describes whether the task completed.
 302    *
 303    * @param bool $isOK
 304    *   TRUE if the task completed successfully.
 305    * @param Exception|NULL $exception
 306    *   If applicable, an unhandled exception that arose during execution.
 307    *
 308    * @return array
 309    *   (is_error => bool, is_continue => bool, numberOfItems => int)
 310    */
 311   public function formatTaskResult($isOK, $exception = NULL) {
 312     $numberOfItems = $this->queue->numberOfItems();
 313
 314     $result = [];
 315     $result['is_error'] = $isOK ? 0 : 1;
 316     $result['exception'] = $exception;
 317     $result['last_task_title'] = $this->lastTaskTitle ?? '';
 318     $result['numberOfItems'] = $this->queue->numberOfItems();
 319     if ($result['numberOfItems'] <= 0) {
 320       // nothing to do
 321       $result['is_continue'] = 0;
 322     }
 323     elseif ($isOK) {
 324       // more tasks remain, and this task succeeded
 325       $result['is_continue'] = 1;
 326     }
 327     elseif ($this->errorMode == CRM_Queue_Runner::ERROR_CONTINUE) {
 328       // more tasks remain, and we can disregard this error
 329       $result['is_continue'] = 1;
 330     }
 331     else {
 332       // more tasks remain, but we can't disregard the error
 333       $result['is_continue'] = 0;
 334     }
 335
 336     return $result;
 337   }
 338
 339   /**
 340    * @return CRM_Queue_TaskContext
 341    */
 342   protected function getTaskContext() {
 343     if (!is_object($this->taskCtx)) {
 344       $this->taskCtx = new CRM_Queue_TaskContext();
 345       $this->taskCtx->queue = $this->queue;
 346       // $this->taskCtx->log = CRM_Core_Config::getLog();
 347       $this->taskCtx->log = CRM_Core_Error::createDebugLogger();
 348     }
 349     return $this->taskCtx;
 350   }
 351
 352 }