CRM/Queue/Runner.php

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