sinopower / work

Work queue library letting you run distributed tasks using a generic abstraction

MyCLabs\Work is a work queue library letting you run tasks in background using a generic abstraction.

It's intent is to be compatible with classic work queue solutions (RabbitMQ, Beanstalkd, …) while offering a high level abstraction.

Current implementations:

• InMemory: synchronous implementation, task are executed directly (useful for tests or dev environments)
• RabbitMQ
• Beanstalkd

Feel free to contribute and submit other implementations (Gearman, …).

Extended guides:

• The InMemory adapter
• The RabbitMQ adapter
• The Beanstalkd adapter
• Listening to events

How it works

In you code (HTTP request for example), you can run a task in background:

$workDispatcher = new RabbitMQWorkDispatcher(/* parameters */);
$workDispatcher->run(new MyTask());

Separately, you set up a worker to run continuously on the command line (like a deamon):

$ php my-worker.php

This worker simply calls:

// my-worker.php
$worker = new RabbitMQWorker(/* parameters */);
// Will loop continuously and execute tasks
$worker->work();

Defining tasks

Define a task:

class BigComputation implements MyCLabs\Work\Task\Task
{
    public $parameter1;
}

And define the code that executes the task:

class BigComputationExecutor implements MyCLabs\Work\TaskExecutor\TaskExecutor
{
    public function execute(Task $task)
    {
        if (! $task instanceof BigComputation) {
            throw new \Exception("Invalid task type provided");
        }
        // Perform the action (here we just multiply the parameter by 2)
        return $task->parameter1 * 2;
    }
}

Execute a task and wait for its result

The run($task) method runs a task in background.

If you want to wait for the result of that task, you have to use a WorkDispatcher that implements the \MyCLabs\Work\Dispatcher\SynchronousWorkDispatcher interface. For example, the RabbitMQ adapter implements this interface.

That interface offers the runAndWait method:

interface SynchronousWorkDispatcher extends WorkDispatcher
{
    /**
     * Run a task in background.
     *
     * You can use $wait to wait a given time for the task to complete.
     * If the task hasn't finished during this time, $timedout will be
     * called and this method will return.
     * If the task has finished, $completed will be called.
     *
     * @param Task     $task
     * @param int      $wait      Number of seconds to wait for the task to complete.
     *                            If 0, doesn't wait.
     * @param callable $completed Called (if $wait > 0) when the task has completed.
     * @param callable $timedout  Called (if $wait > 0) if we hit the timeout while
     *                            waiting.
     * @param callable $errored   Called (if $wait > 0) if the task errors. Takes 1
     *                            parameter which is the exception.
     *
     * @return void No results
     */
    public function runAndWait(
        Task $task,
        $wait = 0,
        callable $completed = null,
        callable $timedout = null,
        callable $errored = null
    );
}

Read more

Read more in the docs.

Contributing

You can run the tests with PHPUnit:

$ composer install
$ vendor/bin/phpunit

Some functional tests need external programs like RabbitMQ or Beanstalkd. For practical reasons, you can boot a VM very quickly using Vagrant and the included configuration. You can then run the tests in the VM:

$ vagrant up
$ vagrant ssh
$ cd /vagrant
$ composer install
$ vendor/bin/phpunit