Tarantool Queue

Tarantool is a NoSQL database and Lua application server. It supports Lua modules, which can be installed with LuaRocks. This package provides PHP bindings for the Tarantool Queue LuaRock.

Table of contents

• Installation
• Before you start
• Working with queue
  • Data types
  • Tasks
  • Producer API
  • Consumer API
  • Statistics
  • Custom methods
• Testing
• License

Installation

The recommended way to install the library is through Composer:

composer require tarantool/queue

Before you start

To use the queue, make sure your Tarantool instance is configured and running. A minimal configuration looks like this:

-- queues.lua

box.cfg {listen = 3301}

queue = require('queue')
queue.create_tube('foobar', 'fifottl', {if_not_exists = true})

You can read more about the box configuration in the official Tarantool documentation. More information on queue configuration can be found here.

To start the instance, copy (or symlink) the queues.lua file into the /etc/tarantool/instances.enabled directory and run the following command:

sudo tt start queues

Working with queue

Once you have your instance running, create a queue object using the queue (tube) name you defined in the Lua script:

use Tarantool\Queue\Queue;

...

$queue = new Queue($client, 'foobar');

where $client is an instance of Tarantool\Client\Client from the tarantool/client package.

Data types

Under the hood, Tarantool uses the MessagePack binary format to serialize and deserialize data stored in a queue. It can handle most PHP data types (except resources and closures) without any manual pre- or post-processing:

$queue->put('foo');
$queue->put(true);
$queue->put(42);
$queue->put(4.2);
$queue->put(['foo' => ['bar' => ['baz' => null]]]);
$queue->put(new MyObject());

To learn more about object serialization, see User-defined types.

Tasks

Most Queue API methods return a Task object that provides the following getters:

Task::getId()
Task::getState() // States::READY, States::TAKEN, States::DONE, States::BURY or States::DELAYED
Task::getData()

It also provides these convenience methods:

Task::isReady()
Task::isTaken()
Task::isDone()
Task::isBuried()
Task::isDelayed()

Producer API

As you've already seen, to insert a task into a queue, call the put() method. It accepts two arguments: the data you want to process and an optional array of task options supported by the queue. For example, the fifottl queue, which we defined earlier in our Lua configuration, supports the delay, ttl, ttr, and pri options:

use Tarantool\Queue\Options;

$queue->put('foo', [Options::DELAY => 30.0]);
$queue->put('bar', [Options::TTL => 5.0]);
$queue->put('baz', [Options::TTR => 10.0, Options::PRI => 42]);

See the full list of available options.

Consumer API

To reserve a task for execution, call the take() method. It accepts an optional timeout parameter. If you supply a timeout, the call waits for up to that many seconds for a READY task to appear in the queue. The method returns either a Task object or null:

$taskOrNull = $queue->take();

// wait 2 seconds
$taskOrNull = $queue->take(2.0);

// wait 100 milliseconds
$taskOrNull = $queue->take(.1);

After successful execution, acknowledge the task to delete it from the queue:

$data = $task->getData();

// process $data

$task = $queue->ack($task->getId());

If the task cannot be executed, return it to the queue:

$task = $queue->release($task->getId());

// For *ttl queues, you can specify a delay.
$task = $queue->release($task->getId(), [Options::DELAY => 30.0]);

To view a task without changing its state, use:

$task = $queue->peek($task->getId());

To bury (disable) a task:

$task = $queue->bury($task->getId());

To return buried tasks to the READY state:

$count = $queue->kick(3); // kick 3 buried tasks

To increase the TTR or TTL of a running task (only for *ttl queues):

$taskOrNull = $queue->touch($takenTask->getId(), 5.0); // increase ttr/ttl to 5 seconds

You can permanently delete a task in any state by calling delete():

$task = $queue->delete($task->getId());

To delete all tasks in a queue:

$queue->truncate();

For a detailed API documentation, please read the section "Using the queue module" of the queue README.

Statistics

The stats() method returns statistics accumulated since the queue was created:

$stats = $queue->stats();

The result of this call might look like this:

[
    'tasks' => [
        'taken'   => 1,
        'buried'  => 1,
        'ready'   => 1,
        'done'    => 0,
        'delayed' => 0,
        'total'   => 3,
    ],
    'calls' => [
        'bury' => 1,
        'put'  => 3,
        'take' => 1,
        ...
    ],
]

In addition, you can specify a key to return only a subset of the array:

$calls = $queue->stats('calls');
$total = $queue->stats('tasks.total');

Custom methods

Thanks to the flexibility of the queue Lua module, you can create your own queue drivers or extend existing ones with additional functionality. For example, suppose you add a put_many method to your foobar queue to insert multiple tasks atomically:

-- queues.lua

...

queue.tube.foobar.put_many = function(self, items)
    local put = {}

    box.begin()
    for k, item in pairs(items) do
        put[k] = tube:put(unpack(item))
    end
    box.commit()

    return put
end

To invoke this method from PHP, use Queue::call():

$result = $queue->call('put_many', [
    'foo' => ['foo', [Options::DELAY => 30.0]],
    'bar' => ['bar'],
]);

Testing

The easiest way to run tests is with Docker. First, build an image using the dockerfile.sh generator:

./dockerfile.sh | docker build -t queue -

Then run a Tarantool instance (needed for integration tests):

docker network create tarantool-php
docker run -d --net=tarantool-php -p 3301:3301 --name=tarantool \
    -v $(pwd)/tests/Integration/queues.lua:/queues.lua \
    tarantool/tarantool:2 tarantool /queues.lua

Then run both unit and integration tests:

docker run --rm --net=tarantool-php -v $(pwd):/queue -w /queue queue

The tests use PHPUnit. You can pass additional arguments and options to the phpunit command. For example, run only the unit tests with:

docker run --rm --net=tarantool-php -v $(pwd):/queue -w /queue queue \
    vendor/bin/phpunit --testsuite=unit

License

The library is released under the MIT License. See the bundled LICENSE file for details.