Tasks

Args

Arguments

Why is the Args pattern complicated?

Because, in version 3.2 of Celery message protocol v.2 defines support for language based task routing. To facilitate this for future use and expansion it become necessary to define argument objects that are language specific.

While support for multiple argument formats is possible, current workers only support Python args.

Basic Args creation with factory

$args = \Rhubarb\Task\Args::newArgs(
    \Rhubarb\Task\Args\Python::LANG,
    2,
    2,
    \Rhubarb\Task\Args\Python\Kwargs::newKwargs(array('args1'=>'val1', 'arg2' => 'val2'))
);

Explicit Python Args using star args and kwargs

$args = new \Rhubarb\Task\Args\Python(
    2,
    2,
    \Rhubarb\Task\Args\Python\Kwargs::newKwargs(array('args1'=>'val1', 'arg2' => 'val2'))
);

Extended RhubarbTaskArgsPythonKwargs usage

Note the object property and array access usage for flexibility.

$kwargs = new \Rhubarb\Task\Args\Python\Kwargs();
$kwargs->arg1 = 'val1';
$kwargs['arg2'] = 'val2';
$args = \Rhubarb\Task\Args::newArgs(null, $kwargs);

Creating your own Arg types

• Creating your own arg types is as simple as creating an object implementing the \Rhubarb\Task\ArgsInterface
• Registering it with the \Rhubarb\Task\Args object
• Fetching it with the factory

/**
 * Class PhpArgs
 * 
 * An hypothetical example class of an Args class for PHP
 */
class PhpArgs implements \Rhubarb\Task\Args\ArgsInterface
{
    const LANG = 'php';
    /**
     * @return mixed
     */
    public function serialize()
    {
        // TODO: Implement serialize() method.
    }

    /**
     * @return string
     */
    public function __toString()
    {
        // TODO: Implement __toString() method.
    }

    /**
     * @return array
     */
    public function toArray()
    {
        // TODO: Implement toArray() method.
    }

    /**
     * @return array
     */
    public function getHeaders()
    {
        // TODO: Implement getHeaders() method.
    }
}
/* Register the new Arg class with \Rhubarb\Task\Args */ 
\Rhubarb\Task\Args::registerType(\PhpArgs::LANG, '\PhpArgs');
$phpArgs = \Rhubarb\Task\Args::newArgs(PhpArgs::LANG);

Signature

Task Signatures

Signatures may be created by calling the \\Rhubarb\\Rhubarb::task method. Providing the name, optional arguments, properties and headers will return a signature. You may use this signature in various ways in your workflow; either as means to call the task or as a template for many tasks.

$sig = $rhubarb->task('tasks.add', Args::newArgs(Python::LANG, 2, 2), array('countdown' => 10));
// tasks.add(2, 2)

$add = $rhubarb->t('task.add');
$add->s(Args::newArgs(Python::LANG, 2, 2));
// tasks.add(2, 2)

This example demonstrates how you may access the properties, args and headers defined within the signature.

$add = $rhubarb->task('task.add', Args::newArgs(Python::LANG, 2, 2, Kwargs::newKwargs(array('arg1' => 1, 'arg2' => 2))));
// tasks.add(2, 2, arg1=1, arg2=2)
$add->getArgs();
// array(
//     'args' => array(1, 2),
//     'kwargs' => array('arg1' => 1, 'arg2' => 2)
// );
$add->getHeaders();
// array(
//     'lang' => 'py',
//     'c_type' => 'tasks.add'
// );

Executing the signature may be down in two ways delay or applyAsync. delay is a wrapper of the other to provide familiarity with the Celery API.

$add = $rhubarb->t('task.add');
$result = $add->applyAsync(Args::newArgs(Python::LANG, 2, 2));
$result->get();
// 4

$add = $rhubarb->t('task.add');
$result = $add->delay(Args::newArgs(Python::LANG, 2, 2), array(), array('countdown' => 1));
$result->get();
// 4

Chains

Task Chains

Creating a task chain Example 1

use Rhubarb\Rhubarb;
use Rhubarb\Task\Args\Python;

$config = include('configuration/predis.php');
$rhubarb = new Rhubarb($config);

$sig = $rhubarb->task('app.add');

$chain = $rhubarb->chain();
for ($i = 0; $i < 10; $i++) {
    $chain->push($sig->s(new Python($i, $i * 10)));
}
$asyncResult = $chain();

Creating a task chain Example 2

use Rhubarb\Rhubarb;
use Rhubarb\Task\Args\Python;

$config = include('configuration/predis.php');
$rhubarb = new Rhubarb($config);
$sig = $rhubarb->task('app.add');

$tasks = array();
for ($i = 0; $i < 10; $i++) {
    $tasks[] = $sig->s(new Python($i, $i * 10));
}
$chain = $rhubarb->chain($tasks);
$asyncResult = $chain();

Groups

Task Groups

TBD

Chords

Task Chords

TBD