README

The WowoQueueBundle provides unified method for use queue systems, like Beanstalkd, RabbitMQ, flat files, database driven queues, etc. For now it only supports Beanstalkd, but you can add your own implementation on your own and send pull request.

Installation

Step 1: Download WowoQueueBundle

If you are using Deps (Symfony 2.0.x)

Add following lines to your deps file:

    [WowoQueueBundle]
        git=git://github.com/wowo/WowoQueueBundle.git
        target=bundles/Wowo/QueueBundle

    [pheanstalk]
        git=https://github.com/pda/pheanstalk
        target=/pheanstalk
        version=v2.1.0

Now, run the vendors script to download the bundle:

$ php bin/vendors install

If you are using Composer (Symfony >= 2.1.x)

Add following lines to your composer.json requirements:

    "require": {
        "wowo/wowo-queue-bundle": "dev-master"
    }

Now, install the bundle with composer:

$ php composer.phar install

Step 2: Configure the Autoloader

(You can jump to Step 3 if you are using composer)

Add the Wowo namespace to your autoloader:

<?php
// app/autoload.php

$loader->registerNamespaces(array(
    // ...
        'Wowo' => __DIR__.'/../vendor/bundles',
        ));

Also add Pheanstalk init on the bottom of autoload:

// ...
require_once __DIR__.'/../vendor/pheanstalk/pheanstalk_init.php';

Step 3: Enable the bundle

Finally, enable the bundle in the kernel:

<?php
// app/AppKernel.php

public function registerBundles()
{
        $bundles = array(
            // ...
            new Wowo\QueueBundle\WowoQueueBundle(),
        );
}

Step 4: install and run beanstalkd

On Debian linux systems (including Ubuntu) you can run:

$ sudo apt-get install beanstalkd

Then run it as a daemon:

$ beanstalkd -d -l 127.0.0.1 -p 11300

Note: If your beanstalkd service is running in other address or port, you must set the following parameter in your configuration:

parameters:
    wowo_queue.default.address: 127.0.0.1:11300

Don't forget to change 127.0.0.1:11300 with your address and port.

Usage

Using Beanstalkd in-memory queueing server

Obtain manager with Beanstalkd implementation from the controller

$manager = $this->get('wowo_queue.manager');

Put some job into the queue to default tube wowo_queue. which can be changed by wowo_queue.pheanstalk.tube parameter

$manager->put(json_encode(array('payload' => 'Hello world', 'date' => new \DateTime())));

Get jobs as soon as they appear in queue. In below example an infinite loop is presented. Such operations should be run as a background tasks, which listens to the queue and processes jobs when they appear.

while ($job = $manager->get()) {
    $data = json_decode($job->getData(), true);
    printf("Job date: %s, payload: %s\n", $data['date']['date'], $data['payload']);
}

Deleting jobs from the queue is fairly easy

$job = $manager->get();
$manager->delete($job);

Using Beanstalkd implementation without service container

Manager object creation is simple and can be done as listed below. This is useful when you don't use bundle in Symfony project.

$implementation = new Wowo\QueueBundle\Implementation\BeanstalkdQueueImplementation();
$implementation->configure(array(
    'address' => 'localhost',
    'tube' => 'foo-tube',
));
$manager = new Wowo\QueueBundle\QueueManager($implementation);

In case you are not using autoloader, Pheanstalk need to be autoloaded by hand:

require_once('pheanstalk/classes/pheanstalk/classloader.php');
\pheanstalk_classloader::register('pheanstalk/classes/');

Amazon SQS

The SQSQueueImplementation utilizes the AWS PHP SDK v2

Configuration

To inject the SQS Implementation into your QueueManager, make sure you instantiate it with a configured SQS client and an associative array mapping tube names to SQS queue names.

Symfony bundle's services.yml example

    # general AWS SQS client
    queue.sqs_client:
        class: Aws\Sqs\SqsClient
        factory_class: Aws\Sqs\SqsClient
        factory_method: factory
        arguments:
          - {credentials: {key: %amazon.s3.key%, secret: %amazon.s3.secret%}, region: %amazon.s3.region%}
    # wowo sqs implementation
    queue.implementation.sqs:
      class: Wowo\QueueBundle\Implementation\SQSQueueImplementation
      arguments: [@queue.sqs_client, %queue.tubes_to_sqs%]

From there, you can simply inject the SQS implementation into your QueueManager to get or put messages

wowo / wowo-queue-bundle

Maintainers

Details