smith-carson / phalcon-queue-db Goto Github PK

$queue->process(function($body, $job) {
  //do stuff with $body or $job
  return true; //true would delete the job, false would bury it, null would release it
});

<?php
use Phalcon\CLI\Task;
use Phalcon\Queue\Db as DbQueue;
use Phalcon\Text;
use PhalconRest\tasks\traits\Options;
use PhalconRest\tasks\workers\Worker;

class WorkTask extends Task
{
    use Options;

    protected static $options     = [];
    protected static $optionsSpec = [
        'limit' => '\d+',
        'delay' => '[\d\.]+',
    ];

    public function __call($action, array $arguments = [])
    {
        $name  = substr($action, 0, -strlen('action'));
        $class = '\PhalconRest\tasks\workers\\'.Text::camelize($name);
        if (class_exists($class)) {
            $options = self::processOptions(isset($arguments[0])? $arguments[0] : []);
            $worker  = new $class($options);

            if (!($worker instanceof Worker)) {
                echo "$class must be an instance of tasks\\workers\\Worker\n";
                die(254);
            }

            $result = (new DbQueue)
                ->watch($name, true)
                ->process([$worker,'handleJob'], static::$options['delay']?: 1, static::$options['limit']);

            echo "\nStats: ".print_r($result, true);
        } else {
            echo "There's no such worker class: $name ($class)\n";
            die(254);
        }
    }

    public function mainAction()
    {
        echo <<<HELP
  -= DbQueue Task runner =-
This scripts walks through available jobs and processes them given a specific class handler.

Usage: app/cli.php work «tube»
Implementation: tasks/workers/«Tube»::handleJob(\$body, Job \$job):bool|void
Options:
    --delay=F   Delay between asking for new jobs when the queue is over
    --limit=I   How much jobs to process before exiting
    --???=???   Other options are passed directly to the Worker constructor

HELP;
    }

}

<?php namespace PhalconRest\tasks\workers;

use Phalcon\Queue\Db\Job;

interface Worker
{
    public function handleJob($body, Job $job);
}

<?php namespace PhalconRest\tasks\traits;

/**
 * Trait Options
 * Adds ability to process options.
 * @static array $optionsSpec Associative array of option name => regex
 *
 * <code>
 *     protected static $optionsSpec = [
 *         'limit'    => '\d*',        //allows --limit=123
 *         'category' => '.*',         //allows --category=u8nreijgndfjn
 *         'fast'     => true/false,   //allows --fast
 *
 * </code>
 */
trait Options
{

    protected static function processOptions(array $args)
    {
        if (!isset(static::$optionsSpec)) {
            return false;
        }

        $spec    = static::$optionsSpec;
        $options = array_combine(array_keys($spec), array_fill(0, sizeof($spec), null));
        foreach (static::$optionsSpec as $name => $regex) {
            $opt   = "--$name";
            $found = false;

            foreach ($args as $k => $arg) {
                if ($regex === true && $arg == $opt) {
                    $options[$name] = true;
                    $found = true;
                } elseif ($regex == '.*' && strpos($arg, $opt.'=')) {
                    $options[$name] = substr($arg, strlen($opt) + 1);
                    $found[] = $arg;
                    $found = true;
                } elseif (preg_match("/$opt=($regex)/", $arg, $matches)) {
                    $options[$name] = $matches[1];
                    $found[] = $arg;
                    $found = true;
                }

                if ($found) {
                    unset($args[$k]); //removes what's found
                    $found = false;
                    break;
                }
            }
        }
        static::$options = $options;

        $rest = [];
        foreach ($args as $arg) {
            $arg    = ltrim($arg, '-');
            $pieces = explode('=', $arg);
            $rest[$pieces[0]] = isset($pieces[1])? $pieces[1] : true;
        }

        return $rest; //returns what's left
    }

}

$queue = new DbQueue();
$queue->bulkFetches = true;
$queue->put(1);
$queue->put(2);
$queue->put(3);
while ($job = $queue->reserve()) { //one query, three jobs
    echo $job->getBody();
    $job->delete(); //one query per job; can't be avoided as further jobs may hang
}