Worker.php

Go to the documentation of this file.

<?php

namespace Rsi\Fred;

class Worker extends Component{

  public $classNameAlias = []; //!< Job class name aliasses (key = alias, value = class name).
  public $timeLimit = 60; //!< Default time limit for a job (seconds).
  public $breakCount = 100; //!< Halt the processig of job after this amount (even when there are more jobs scheduled).
  public $breakTime = 300; //!< Halt the processig of job after this time (seconds; even when there are more jobs scheduled).
  public $defaultOverduePeriod = '+1 hour'; //!< Default overdue period for jobs that do not have a rescheduling interval.
  public $logTimerPrio = null; //!< Log the begin and end times of a job with this (custom) prio.

  /**
   * Translate a job class name.
   * @param string $class_name  Class name for the job, or an alias.
   * @return string  Class name.
   */
  protected function transClassName($class_name){
    return $this->classNameAlias[$class_name] ?? $class_name;
  }
  /**
   * Add a job to the queue.
   * @param string $class_name  Class name for the job.
   * @param array $config  Configuration options (key-value pairs) for the job.
   * @param int $time  Time to run the job (null for as soon as possible).
   * @param string $interval  String that is understood by PHP's strtotime() to reschedule the job after it is finished.
   * @return string  Job ID when successful.
   */
  public function schedule($class_name,$config = null,$time = null,$interval = null){
    return false;
  }
  /**
   * Delete a scheduled job from the queue.
   * @param string $id  Job ID.
   * @return bool  True when successful.
   */
  public function delete($id){
    return false;
  }
  /**
   * Create a job object.
   * @param string $class_name  Class name for the job.
   * @param string $id  Unique ID for the job.
   * @param array $config  Configuration options (key-value pairs) for the job.
   * @param int $time  Time to run the job (null for as soon as possible).
   * @param string $interval  String that is understood by PHP's strtotime() to reschedule the job after it is finished.
   * @return \\Rsi\\Fred\\Worker\\Job
   */
  protected function create($class_name,$id,$config = null,$time = null,$interval = null){
    return new $class_name($this,$id,$config,$time,$interval);
  }
  /**
   * Run a job instantly.
   * @param string $class_name  Class name for the job.
   * @param array $config  Configuration options (key-value pairs) for the job.
   */
  public function instant($class_name,$config = null){
    $this->create($class_name,null,$config)->execute();
  }
  /**
   * Next scheduled job to execute.
   * @param string $id  Specific job ID to execute (whether scheduled time has come or not).
   * @return \\Rsi\\Fred\\Worker\\Job (empty if there is no job to be executed).
   */
  protected function next($id = null){
    return null;
  }
  /**
   * Lock a job.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job to lock.
   * @return bool  True if the job has been locked.
   */
  protected function lock($job){
    return false;
  }
  /**
   * Unlock a job.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job to unlock.
   * @return bool  True if the job has been unlocked.
   */
  protected function unlock($job){
    return true;
  }
  /**
   * Reschedule a job.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job to reschedule.
   * @param int $time  Time to reschedule at (null for as soon as possible).
   * @return bool  True if the job has been rescheduled.
   */
  protected function reschedule($job,$time = null){
    return false;
  }
  /**
   * Mark a job as done.
   * @param \\Rsi\\Fred\\Worker\\Job $job  Job.
   */
  protected function ready($job){
    $this->delete($job->id);
  }
  /**
   * Overdue jobs.
   * Jobs that have been locked longer than twice their interval or the default overdue interval.
   * @return array  Of \\Rsi\\Fred\\Worker\\Job items.
   */
  protected function overdue(){
    return [];
  }
  /**
   * Execute the worker / process the scheduled jobs.
   * Call his function as often as possible.
   * @param string $id  Specific job ID to execute (whether scheduled time has come or not).
   */
  public function execute($id = null){
    $log = $this->component('log');
    $count = 0;
    $start = time();
    while(($job = $this->next($id)) && $this->lock($job)) try{
      $this->_fred->timeLimit = $this->timeLimit;
      $timer_id = $this->logTimerPrio ? $log->start($this->logTimerPrio,"Executing job '{$job->id}'",__FILE__,__LINE__) : null;
      $job->execute();
      if($timer_id) $log->stop($timer_id);
      if($job->interval){
        $diff = ($time = $job->nextTime) - time();
        if(!$time) $log->alert("Invalid interval for job '{$job->id}'");
        elseif($diff < 0) $log->warning("Job '{$job->id}' rescheduled in past",['diff' => $diff]);
        $this->reschedule($job,$time);
      }
      else $this->ready($job);
      $this->unlock($job);
      if((++$count >= $this->breakCount) || (time() - $start >= $this->breakTime)) break;
    }
    catch(\Exception $e){
      $log->critical($e);
    }
    foreach($this->overdue() as $job)
      $log->critical("Job '{$job->id}' is overdue",['time' => $job->time,'interval' => $job->interval]);
  }

}