158 lines
3.5 KiB
PHP
158 lines
3.5 KiB
PHP
<?php
|
|
|
|
abstract class Job {
|
|
private $_jobid = 0;
|
|
private $_source = "unknown";
|
|
private $_owner = "system";
|
|
public $retries = 0;
|
|
|
|
public function __construct($source = "unknown") {
|
|
$this->_source = $source;
|
|
}
|
|
|
|
// Implement this function to actually do the job
|
|
abstract public function run();
|
|
|
|
// Place the current job onto the queue ready for processing.
|
|
public function queue($owner = -1) {
|
|
$class = get_called_class();
|
|
$data = serialize($this);
|
|
$db = DB::getInstance();
|
|
|
|
if ($owner == -1) {
|
|
$user = get_user();
|
|
if ($user) {
|
|
$owner = $user->id;
|
|
}
|
|
}
|
|
|
|
$this->_jobid = $db->insert("job", [
|
|
"class" => $class,
|
|
"data" => $data,
|
|
"queued" => time(),
|
|
"source" => $this->_source,
|
|
"status" => "Queued",
|
|
"owner" => $owner
|
|
]);
|
|
|
|
return $this->_jobid;
|
|
}
|
|
|
|
public function getJobClass() {
|
|
$class = get_called_class();
|
|
return $class;
|
|
}
|
|
|
|
public function jobID() {
|
|
return $this->_jobid;
|
|
}
|
|
|
|
public function setJobID($id) {
|
|
$this->_jobid = $id;
|
|
}
|
|
|
|
// Set the current status message
|
|
public function status($txt) {
|
|
$db = DB::getInstance();
|
|
$db->update("job", $this->_jobid, [
|
|
"status" => substr($txt, 0, 200)
|
|
]);
|
|
if (is_terminal()) {
|
|
print(" => " . $txt . "\n");
|
|
}
|
|
}
|
|
|
|
// Mark the job as completed successfully
|
|
public function finish() {
|
|
$db = DB::getInstance();
|
|
$db->update("job", $this->_jobid, [
|
|
"finished" => time(),
|
|
]);
|
|
}
|
|
|
|
// Mark the job as failed miserably
|
|
public function fail($message = "") {
|
|
$db = DB::getInstance();
|
|
if ($message != "") {
|
|
$db->update("job", $this->_jobid, [
|
|
"status" => substr($message, 0, 200),
|
|
"failed" => time(),
|
|
]);
|
|
} else {
|
|
$db->update("job", $this->_jobid, [
|
|
"failed" => time(),
|
|
]);
|
|
}
|
|
}
|
|
|
|
// Restart the job.
|
|
public function retry($msg = "", $when = -1) {
|
|
if ($when == -1) {
|
|
$when = rand(60, 600);
|
|
}
|
|
$db = DB::getInstance();
|
|
$status = "Retrying " . date("Y-m-d H:i", time() + $when);
|
|
if ($msg != "") {
|
|
$status .= " ($msg)";
|
|
}
|
|
|
|
$db->update("job", $this->_jobid, [
|
|
"failed" => 0,
|
|
"finished" => 0,
|
|
"started" => 0,
|
|
"status" => $status,
|
|
"retry" => time() + $when
|
|
]);
|
|
}
|
|
|
|
// Restart the job and defer it to a later time.
|
|
public function defer($when) {
|
|
$db = DB::getInstance();
|
|
$db->update("job", $this->_jobid, [
|
|
"failed" => 0,
|
|
"finished" => 0,
|
|
"started" => 0,
|
|
"queued" => $when,
|
|
"status" => "Deferred until " . gmdate("Y-m-d\TH:i:s\Z", $when)
|
|
]);
|
|
}
|
|
|
|
// Look for the next available job that optionally has
|
|
// the requested class. Mark it as started, deserialize
|
|
// it, and return the job runner object.
|
|
// Returns false if no job available.
|
|
public static function consumeNextJob($class = null) {
|
|
$db = DB::getInstance();
|
|
|
|
$db->query("lock table job write");
|
|
if ($class == null) {
|
|
$q = $db->query("select * from job where started=0 and (retry=0 or retry < unix_timestamp(now())) and queued < unix_timestamp(now()) order by queued limit 1");
|
|
} else {
|
|
$q = $db->query("select * from job where started=0 and (retry=0 or retry < unix_timestamp(now())) and queued < unix_timestamp(now()) and class=:class order by queued limit 1", ["class" => $class]);
|
|
}
|
|
$r = $db->nextRecord($q);
|
|
if (!$r) {
|
|
$db->query("unlock tables");
|
|
return false;
|
|
}
|
|
|
|
$db->update("job", $r->id, [
|
|
"started" => time()
|
|
]);
|
|
$db->query("unlock tables");
|
|
|
|
$ob = unserialize($r->data);
|
|
$ob->setJobID($r->id);
|
|
$ob->setOwner($r->owner);
|
|
return $ob;
|
|
}
|
|
|
|
public function getOwner() {
|
|
return $this->_owner;
|
|
}
|
|
|
|
public function setOwner($owner) {
|
|
$this->_owner = $owner;
|
|
}
|
|
}
|