DJJob.php

<?php

# This system is mostly a port of delayed_job: http://github.com/tobi/delayed_job

class DJException extends Exception { }

/**
 * Exception thrown when the job should be retried after a specific period.
 */
class DJRetryException extends DJException {

    private $delay_seconds = 7200;

    public function setDelay($delay) {
        $this->delay_seconds = $delay;
    }
    public function getDelay() {
        return $this->delay_seconds;
    }
}

/**
 * Job handler interface which can be safely enqueued.
 */
interface DJJobHandlerInterface {
    /**
     * Method that will perform the job when retrieved from the jobs table.
     */
    public function perform();
}

/**
 * Base class for delayed job.
 */
class DJBase {

    // error severity levels
    const CRITICAL = 4;
    const    ERROR = 3;
    const     WARN = 2;
    const     INFO = 1;
    const    DEBUG = 0;

    /**
     * @var int
     */
    private static $log_level = self::DEBUG;

    /**
     * @var null|PDO
     */
    private static $db = null;

    /**
     * @var string
     */
    protected static $jobsTable = "";

    /**
     * @var string
     */
    private static $dsn = "";

    /**
     * @var string
     */
    private static $user = "";

    /**
     * @var string
     */
    private static $password = "";

    /**
     * @var int
     */
    private static $retries = 3; //default retries

    // use either `configure` or `setConnection`, depending on if
    // you already have a PDO object you can re-use

    /**
     * Configures DJJob with certain values for the database connection.
     */
    public static function configure(){
        $args = func_get_args();
        $numArgs = func_num_args();

        switch ($numArgs) {
            case 1:{
                if (is_array($args[0])){
                    self::configureWithOptions($args[0]);
                } else {
                    self::configureWithDsnAndOptions($args[0]);
                }
                break;
            }
            case 2:{
                if (is_array($args[0])){
                    self::configureWithOptions($args[0], $args[1]);
                } else {
                    self::configureWithDsnAndOptions($args[0], $args[1]);
                }
                break;
            }
            case 3: {
                self::configureWithDsnAndOptions($args[0], $args[1], $args[2]);
                break;
            }
        }
    }

    /**
     * Configures DJJob with certain values for the database connection.
     *
     * @param        $dsn The PDO connection string.
     * @param array  $options The options for the PDO connection.
     * @param string $jobsTable Name of the jobs table.
     *
     * @throws DJException Throws an exception with invalid parameters.
     */
    protected static function configureWithDsnAndOptions($dsn, array $options = array(), $jobsTable = 'jobs') {
        if (!isset($options['mysql_user'])){
            throw new DJException("Please provide the database user in configure options array.");
        }
        if (!isset($options['mysql_pass'])){
            throw new DJException("Please provide the database password in configure options array.");
        }

        self::$dsn = $dsn;
        self::$jobsTable = $jobsTable;

        self::$user = $options['mysql_user'];
        self::$password = $options['mysql_pass'];

        // searches for retries
        if (isset($options['retries'])){
            self::$retries = (int) $options['retries'];
        }
    }

    /**
     * @param array  $options
     * @param string $jobsTable
     *
     * @throws DJException Throws an exception with invalid parameters.
     */
    protected static function configureWithOptions(array $options, $jobsTable = 'jobs') {

        if (!isset($options['driver'])){
            throw new DJException("Please provide the database driver used in configure options array.");
        }
        if (!isset($options['user'])){
            throw new DJException("Please provide the database user in configure options array.");
        }
        if (!isset($options['password'])){
            throw new DJException("Please provide the database password in configure options array.");
        }

        self::$user = $options['user'];
        self::$password = $options['password'];
        self::$jobsTable = $jobsTable;

        self::$dsn = $options['driver'] . ':';
        foreach ($options as $key => $value) {
            // skips options already used
            if ($key == 'driver' || $key == 'user' || $key == 'password') {
                continue;
            }

            // searches for retries
            if ($key == 'retries'){
                self::$retries = (int) $value;
                continue;
            }

            self::$dsn .= $key . '=' . $value . ';';
        }
    }

    /**
     * @param int $const The log level to set.
     */
    public static function setLogLevel($const) {
        self::$log_level = $const;
    }

    /**
     * @param PDO $db The database connection to use.
     */
    public static function setConnection(PDO $db) {
        self::$db = $db;
    }

    /**
     * Returns the connection DJBase knows about.
     * 
     * Tries to connect if no connection is present.
     * 
     * @return null|PDO The connection if a valid connection is present.
     * @throws Exception
     */
    protected static function getConnection() {
        if (self::$db === null) {
            try {
                self::$db = new PDO(self::$dsn, self::$user, self::$password);
                self::$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
            } catch (PDOException $e) {
                throw new Exception("DJJob couldn't connect to the database. PDO said [{$e->getMessage()}]");
            }
        }
        return self::$db;
    }

    /**
     * Runs a query with a resultset against the database.
     * 
     * @param string $sql The query to execute.
     * @param array  $params The params necessary for a prepared statement.
     *
     * @return array Returns the complete resultset.
     * @throws DJException Throws if the query couldn't be executed.
     */
    public static function runQuery($sql, $params = array()) {
        for ($attempts = 0; $attempts < self::$retries; $attempts++) {
            try {
                $stmt = self::getConnection()->prepare($sql);
                $stmt->execute($params);

                $ret = array();
                if ($stmt->rowCount()) {
                    // calling fetchAll on a result set with no rows throws a
                    // "general error" exception
                    foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $r) $ret []= $r;
                }

                $stmt->closeCursor();
                return $ret;
            }
            catch (PDOException $e) {
                // Catch "MySQL server has gone away" error.
                if ($e->errorInfo[1] == 2006) {
                    self::$db = null;
                }
                // Throw all other errors as expected.
                else {
                    throw $e;
                }
            }
        }

        throw new DJException("DJJob exhausted retries connecting to database");
    }

    /**
     * Runs an update query against the database.
     *
     * @param string $sql The query to execute.
     * @param array $params The params necessary for the prepared statement.
     *
     * @return int The amount of affected rows.
     * @throws DJException Throws if the query couldn't be executed.
     */
    public static function runUpdate($sql, $params = array()) {
        for ($attempts = 0; $attempts < self::$retries; $attempts++) {
            try {
                $stmt = self::getConnection()->prepare($sql);
                $stmt->execute($params);
                return $stmt->rowCount();
            }
            catch (PDOException $e) {
                // Catch "MySQL server has gone away" error.
                if ($e->errorInfo[1] == 2006) {
                    self::$db = null;
                }
                // Throw all other errors as expected.
                else {
                    throw $e;
                }
            }
        }

        throw new DJException("DJJob exhausted retries connecting to database");
    }

    /**
     * Logs a message to the output.
     *
     * @param string $mesg The message to log.
     * @param int $severity The log level necessary for this message to display.
     */
    protected static function log($mesg, $severity=self::CRITICAL) {
        if ($severity >= self::$log_level) {
            printf("[%s] %s\n", date('c'), $mesg);
        }
    }
}

/**
 * The worker class that can empty a queue.
 */
class DJWorker extends DJBase {
    # This is a singleton-ish thing. It wouldn't really make sense to
    # instantiate more than one in a single request (or commandline task)

    /**
     * DJWorker constructor.
     *
     * The following options are available:
     * `queue`: The queue to work on. Default: 'default'
     * `count`: How many jobs to execute before exiting. Use '0' for no-limit. Default: '0'
     * `sleep`: How long to sleep if no jobs are found. Default: '5'
     * `max_attempts`: How many times to try a job before bailing out. Default: '5'
     * `fail_on_output`: Whether to fail on output. Default: 'false'
     *
     * @param array $options The settings for this worker.
     */
    public function __construct($options = array()) {
        $options = array_merge(array(
            "queue" => "default",
            "count" => 0,
            "sleep" => 5,
            "max_attempts" => 5,
            "fail_on_output" => false
        ), $options);
        list($this->queue, $this->count, $this->sleep, $this->max_attempts, $this->fail_on_output) =
            array($options["queue"], $options["count"], $options["sleep"], $options["max_attempts"], $options["fail_on_output"]);

        list($hostname, $pid) = array(trim(gethostname()), getmypid());
        $this->name = "host::$hostname pid::$pid";

        if (function_exists("pcntl_signal")) {
            pcntl_signal(SIGTERM, array($this, "handleSignal"));
            pcntl_signal(SIGINT, array($this, "handleSignal"));
        }
    }

    /**
     * Handles a signal from the operating system.
     *
     * @param string $signo The signal received from the OS.
     */
    public function handleSignal($signo) {
        $signals = array(
            SIGTERM => "SIGTERM",
            SIGINT  => "SIGINT"
        );
        $signal = $signals[$signo];

        $this->log("[WORKER] Received received {$signal}... Shutting down", self::INFO);
        $this->releaseLocks();
        die(0);
    }

    /**
     * Releases all locks this worker has on the jobs table.
     */
    public function releaseLocks() {
        $this->runUpdate("
            UPDATE " . self::$jobsTable . "
            SET locked_at = NULL, locked_by = NULL
            WHERE locked_by = ?",
            array($this->name)
        );
    }

    /**
     * Returns a new job ordered by most recent first
     * why this?
     *     run newest first, some jobs get left behind
     *     run oldest first, all jobs get left behind
     *
     * @return \DJJob|false A job if one was successfully locked. Otherwise false.
     */
    public function getNewJob() {
        # we can grab a locked job if we own the lock
        $rs = $this->runQuery("
            SELECT id
            FROM   " . self::$jobsTable . "
            WHERE  queue = ?
            AND    (run_at IS NULL OR NOW() >= run_at)
            AND    (locked_at IS NULL OR locked_by = ?)
            AND    failed_at IS NULL
            AND    attempts < ?
            ORDER BY created_at DESC
            LIMIT  10
        ", array($this->queue, $this->name, $this->max_attempts));

        // randomly order the 10 to prevent lock contention among workers
        shuffle($rs);

        foreach ($rs as $r) {
            $job = new DJJob($this->name, $r["id"], array(
                "max_attempts" => $this->max_attempts,
                "fail_on_output" => $this->fail_on_output
            ));
            if ($job->acquireLock()) return $job;
        }

        return false;
    }

    /**
     * Starts the worker process.
     */
    public function start() {
        $this->log("[JOB] Starting worker {$this->name} on queue::{$this->queue}", self::INFO);

        $count = 0;
        $job_count = 0;
        try {
            while ($this->count == 0 || $count < $this->count) {
                if (function_exists("pcntl_signal_dispatch")) pcntl_signal_dispatch();

                $count += 1;
                $job = $this->getNewJob($this->queue);

                if (!$job) {
                    $this->log("[JOB] Failed to get a job, queue::{$this->queue} may be empty", self::DEBUG);
                    sleep($this->sleep);
                    continue;
                }

                $job_count += 1;
                $job->run();
            }
        } catch (Exception $e) {
            $this->log("[JOB] unhandled exception::\"{$e->getMessage()}\"", self::ERROR);
        }

        $this->log("[JOB] worker shutting down after running {$job_count} jobs, over {$count} polling iterations", self::INFO);
    }
}

/**
 * Represents a job that needs to be executed.
 */
class DJJob extends DJBase {

    /**
     * Constructs the Job
     *
     * Possible options:
     * `max_attempts`: The amount of attempts before bailing out. Default: '5'
     * `fail_on_output`: Whether the job fails if there is output in the handler. Default: 'false'
     *
     * @param string $worker_name Name of the worker that created this job.
     * @param int $job_id ID of this job.
     * @param array $options The options.
     */
    public function __construct($worker_name, $job_id, $options = array()) {
        $options = array_merge(array(
            "max_attempts" => 5,
            "fail_on_output" => false
        ), $options);
        $this->worker_name = $worker_name;
        $this->job_id = $job_id;
        $this->max_attempts = $options["max_attempts"];
        $this->fail_on_output = $options["fail_on_output"];
    }

    /**
     * Runs this job.
     *
     * First retrieves the handler fro the database. Then perform the job.
     *
     * @return bool Whether or not the job succeeded.
     */
    public function run() {
        # pull the handler from the db
        $handler = $this->getHandler();
        if (!is_object($handler)) {
            $msg = "[JOB] bad handler for job::{$this->job_id}";
            $this->finishWithError($msg);
            return false;
        }

        # run the handler
        try {

            if ($this->fail_on_output) {
                ob_start();                
            }

            $handler->perform();

            if ($this->fail_on_output) {
                $output = ob_get_contents();
                ob_end_clean();

                if (!empty($output)) {
                    throw new Exception("Job produced unexpected output: $output");
                }
            }

            # cleanup
            $this->finish();
            return true;

        } catch (DJRetryException $e) {
            if ($this->fail_on_output) {
                ob_end_flush();
            }
            
            # attempts hasn't been incremented yet.
            $attempts = $this->getAttempts()+1;

            $msg = "Caught DJRetryException \"{$e->getMessage()}\" on attempt $attempts/{$this->max_attempts}.";

            if($attempts == $this->max_attempts) {
                $msg = "[JOB] job::{$this->job_id} $msg Giving up.";
                $this->finishWithError($msg, $handler);
            } else {
                $this->log("[JOB] job::{$this->job_id} $msg Try again in {$e->getDelay()} seconds.", self::WARN);
                $this->retryLater($e->getDelay());
            }
            return false;

        } catch (Exception $e) {
            if ($this->fail_on_output) {
                ob_end_flush();
            }

            $this->finishWithError($e->getMessage(), $handler);
            return false;

        }
    }

    /**
     * Acquires lock on this job.
     *
     * @return bool Whether or not acquiring the lock succeeded.
     */
    public function acquireLock() {
        $this->log("[JOB] attempting to acquire lock for job::{$this->job_id} on {$this->worker_name}", self::INFO);

        $lock = $this->runUpdate("
            UPDATE " . self::$jobsTable . "
            SET    locked_at = NOW(), locked_by = ?
            WHERE  id = ? AND (locked_at IS NULL OR locked_by = ?) AND failed_at IS NULL
        ", array($this->worker_name, $this->job_id, $this->worker_name));

        if (!$lock) {
            $this->log("[JOB] failed to acquire lock for job::{$this->job_id}", self::INFO);
            return false;
        }

        return true;
    }

    /**
     * Releases the lock on this job.
     */
    public function releaseLock() {
        $this->runUpdate("
            UPDATE " . self::$jobsTable . "
            SET locked_at = NULL, locked_by = NULL
            WHERE id = ?",
            array($this->job_id)
        );
    }

    /**
     * Finishes this job. Will delete it from the jobs table.
     */
    public function finish() {
        $this->runUpdate(
            "DELETE FROM " . self::$jobsTable . " WHERE id = ?",
            array($this->job_id)
        );
        $this->log("[JOB] completed job::{$this->job_id}", self::INFO);
    }

    /**
     * Finishes this job, but with an error. Keeps the job in the jobs table.
     *
     * @param string $error The error message to write to the job.
     * @param null|object $handler The handler that ran this job.
     */
    public function finishWithError($error, $handler = null) {
        $this->runUpdate("
            UPDATE " . self::$jobsTable . "
            SET attempts = attempts + 1,
                failed_at = IF(attempts >= ?, NOW(), NULL),
                error = IF(attempts >= ?, ?, NULL)
            WHERE id = ?",
            array(
                $this->max_attempts,
                $this->max_attempts,
                $error,
                $this->job_id
            )
        );
        $this->log($error, self::ERROR);
        $this->log("[JOB] failure in job::{$this->job_id}", self::ERROR);
        $this->releaseLock();

        if ($handler && ($this->getAttempts() == $this->max_attempts) && method_exists($handler, '_onDjjobRetryError')) {
          $handler->_onDjjobRetryError($error);
        }
    }

    /**
     * Saves a retry date to this job.
     *
     * @param int $delay The amount of seconds to delay this job.
     */
    public function retryLater($delay) {
        $this->runUpdate("
            UPDATE " . self::$jobsTable . "
            SET run_at = DATE_ADD(NOW(), INTERVAL ? SECOND),
                attempts = attempts + 1
            WHERE id = ?",
            array(
              $delay,
              $this->job_id
            )
        );
        $this->releaseLock();
    }

    /**
     * Returns the handler for this job.
     *
     * @return bool|object The handler object for this job. Or false if it failed.
     */
    public function getHandler() {
        $rs = $this->runQuery(
            "SELECT handler FROM " . self::$jobsTable . " WHERE id = ?",
            array($this->job_id)
        );
        foreach ($rs as $r) return unserialize($r["handler"]);
        return false;
    }

    /**
     * Returns the amount of attempts left for this job.
     *
     * @return bool The amount of attempts left.
     */
    public function getAttempts() {
        $rs = $this->runQuery(
            "SELECT attempts FROM " . self::$jobsTable . " WHERE id = ?",
            array($this->job_id)
        );
        foreach ($rs as $r) return $r["attempts"];
        return false;
    }

    /**
     * Enqueues a job to the database.
     *
     * @param object $handler The handler that can execute this job.
     * @param string $queue The queue to enqueue this job to. All queues are saved in the same table.
     * @param string $run_at A valid mysql DATETIME string at which to run the jobs.
     *
     * @return bool|string Returns the last inserted ID or false if enqueuing failed.
     */
    public static function enqueue($handler, $queue = "default", $run_at = null) {
        $affected = self::runUpdate(
            "INSERT INTO " . self::$jobsTable . " (handler, queue, run_at, created_at) VALUES(?, ?, ?, NOW())",
            array(serialize($handler), (string) $queue, $run_at)
        );

        if ($affected < 1) {
            self::log("[JOB] failed to enqueue new job", self::ERROR);
            return false;
        }

        return self::getConnection()->lastInsertId(); // return the job ID, for manipulation later
    }

    /**
     * Bulk enqueues a lot of jobs to the database.
     *
     * @param object[] $handlers An array of handlers to enqueue.
     * @param string $queue The queue to enqueue the handlers to.
     * @param string $run_at A valid mysql DATETIME string at which to run the jobs.
     *
     * @return bool
     */
    public static function bulkEnqueue($handlers, $queue = "default", $run_at = null) {
        $sql = "INSERT INTO " . self::$jobsTable . " (handler, queue, run_at, created_at) VALUES";
        $sql .= implode(",", array_fill(0, count($handlers), "(?, ?, ?, NOW())"));

        $parameters = array();
        foreach ($handlers as $handler) {
            $parameters []= serialize($handler);
            $parameters []= (string) $queue;
            $parameters []= $run_at;
        }
        $affected = self::runUpdate($sql, $parameters);

        if ($affected < 1) {
            self::log("[JOB] failed to enqueue new jobs", self::ERROR);
            return false;
        }

        if ($affected != count($handlers))
            self::log("[JOB] failed to enqueue some new jobs", self::ERROR);

        return true;
    }

    /**
     * Returns the general status of the jobs table.
     *
     * @param string $queue The queue of which to see the status for.
     *
     * @return array Information about the status.
     */
    public static function status($queue = "default") {
        $rs = self::runQuery("
            SELECT COUNT(*) as total, COUNT(failed_at) as failed, COUNT(locked_at) as locked
            FROM `" . self::$jobsTable . "`
            WHERE queue = ?
        ", array($queue));
        $rs = $rs[0];

        $failed = $rs["failed"];
        $locked = $rs["locked"];
        $total  = $rs["total"];
        $outstanding = $total - $locked - $failed;

        return array(
            "outstanding" => $outstanding,
            "locked" => $locked,
            "failed" => $failed,
            "total"  => $total
        );
    }

}