Presentation

Beanstalkd is a message queue server, which on the one hand registers processing orders, and on the other hand retrieves these orders for processing.

If you have correctly configured the Beanstalkd connection parameters, Temma automatically creates an object of type \Temma\Datasources\Beanstalk. By convention, we'll assume that you've named this connection beanstalk in the temma.json file (see configuration documentation).

The connection is then available in the controller by writing:

$this->beanstalk

In other objects managed by the dependency injection component, the Beanstalkd connection is accessible by writing:

$loader->dataSources['beanstalk']

Installation

To connect to a Beanstalk server, you need to install the Pheanstalk library. To do this, you can use the Composer dependency manager.

At the root of the project, create the following composer.json file:

{
    "require": {
        "pda/pheanstalk": "v5.x-dev"
    }
}

Next, run the following command: composer update

Configuration

In the temma.json file (see configuration documentation), you declare the DSN (Data Source Name) used to connect to the Beanstalkd server.

The DSN used to connect to the Beanstalkd server is written as: beanstalk://SERVER[:PORT]/TUBE_NAME
The default port number is 11300.
Example: beanstalk://localhost/tube3

Specific features of message queues

Unlike other data sources, messages are added to a message queue without providing an identifier (name, key, path, etc.). The identifier is created automatically by Beanstalkd. When a message is read, its identifier is returned along with its data; this identifier is subsequently used to delete the message from the queue.

Consequently, Beanstalkd does not allow you to check the existence of a message based on its identifier, nor to select a message, nor to define a default value when retrieving a message.

Array-type access

// write a message (serialized)
$this->beanstalk[''] = $data;

// read a message (deserialized)
$msg = $this->beanstalk[''];
// returns an associative array with the "id" and "data" keys

// delete a message
unset($this->beanstalk[$msg['id']]);

// waiting messages count
$nbr = count($this->beanstalk);

General methods

// delete a message
$this->beanstalk->remove($msg['id']);

Management of complex serialized data

// read a message (deserialized)
$msg = $this->beanstalk->get('');
// returns an associative array with the "id" and "data" keys

// write a message (serialized)
$this->beanstalk->set('', $msgData);

// write multiple messages (serialized)
$this->beanstalk->mSet([
    $user1data,
    $user2data,
    $user3data,
]);

Raw data management

// read a message (raw)
$html = $this->beanstalk->read('');
// returns an associative array with the "id" and "data" keys

// write a message (raw)
$this->beanstalk->write('', $msgData);

// write multiple messages (raw)
$this->beanstalk->mWrite({
    $user1data,
    $user2data,
    $user3data,
]);

// write a message (raw) from a local file
$this->beanstalk->copyTo('', '/path/to/file');

// write multiple messages (raw) from local files
$this->beanstalk->mCopyTo([
    '/path/to/file1',
    '/path/to/file2',
    '/path/to/file3',
]);

touch

touch(string $id) : \Temma\Datasources\Beanstalk

This method tells the Beanstalkd server that the message is still being processed, thus preventing it from being distributed again.
Example:

// retrieve the next message to be processed
$msg = $this->beanstalk->get('');

// processing...

// tells the server that processing is still in progress
$this->beanstalk->touch($msg['id']);

// further processing...

// delete the message
$this->beanstalk->remove($msg['id']);