QueueBundle

Introduction
Creating Jobs
- Generating Job Classes
- Class Structure
Dispatching Jobs
Running The Queue Worker
Supervisor Configuration
Dealing With Failed Jobs
- Cleaning Up After Failed Jobs
- Retrying Failed Jobs
Job Events
Other Database and library support
GIST

Introduction

This QueueBundle is heavily inspired by Laravel Queue package. In fact some of the file are directly copied over. So, hats off to Taylor Otwell and Laravel team for providing an awesome package for the community.

Installation

You can install QueueBundle by composer

composer require ideasbucket/queue-bundle

QueueBundle supports Symfony 2.8, 3.0 and above.

Configuration

Once you install the bundle you will need to make change in your AppKernel.php by adding the bundle class entry like this.

<?php

use Symfony\Component\HttpKernel\Kernel;
use Symfony\Component\Config\Loader\LoaderInterface;

class AppKernel extends Kernel
{
    public function registerBundles()
    {
        $bundles = array(
            new Symfony\Bundle\FrameworkBundle\FrameworkBundle(),
            new Symfony\Bundle\SecurityBundle\SecurityBundle(),
            new Symfony\Bundle\TwigBundle\TwigBundle(),
            new Symfony\Bundle\MonologBundle\MonologBundle(),
            new Symfony\Bundle\SwiftmailerBundle\SwiftmailerBundle(),
            new Doctrine\Bundle\DoctrineBundle\DoctrineBundle(),
            new Sensio\Bundle\FrameworkExtraBundle\SensioFrameworkExtraBundle(),
            new IdeasBucket\QueueBundle\IdeasBucketQueueBundle(), // ADD THIS
            new AppBundle\AppBundle(),
        );

        ....

        return $bundles;
    }
}

Then in config.yml you can define configurations for each of the queue drivers that are included with the bundle, which includes a database, Beanstalkd, Amazon SQS, Redis, and a synchronous driver that will execute jobs immediately (for local use). A null queue driver is also included which simply discards queued jobs.

Basic minimal configuration that is needed for QueueBundle is to configure cache_handler. Basically it can be any service that implements any one of these interfaces.

PSR-16 Cache
PSR-6 Cache Pool
Doctrine Cache

For cache handler you can define service like this.

# In service.yml or config.yml file
app.cache:
    app: ANY_CACHE_ADAPTER

If possible we recommend PSR-16 cache interface.

You can use any cache adapter here. For more information regarding cache handler please visit here or here .

Full configuration for QueueBundle is following.

ideasbucket_queue:
    cache_handler: app.cache
    default: sync # default
    # Default config for command path you may need to change 
    # this if you are using Symfony 2.x directory structure.
    command_path: %kernel.root_dir%/../bin/
    lock_path: ~
    lock_service: ideasbucket_queue.filesystem_switch # Default value
    connections:
        sqs:
            driver: sqs
            key: YOUR_KEY
            secret: YOUR_SECRET
            prefix: https://sqs.us-west-2.amazonaws.com/some-id
            queue: default
            region: us-west-2
        redis:
            driver: redis
            client: YOUR_PREDIS_CLIENT
            queue: default
            retry_after: 90
        beanstalkd:
            driver: beanstalkd
            host: localhost
            port: 11300
            persistent: ~
            queue: default
            retry_after: 90
        database:
            driver: database
            queue: default
            repository: YOUR_QUEUE_REPOSITORY
            retry_after: 90
    # If you want to store failed jobs in database.          
    #failed_job_repository: FAILED_JOB_REPOSITORY

Connections vs. Queues

Before getting started with QueueBundle, it is important to understand the distinction between "connections" and "queues". In your config.yml you can define configuration for connections. This option defines a particular connection to a backend service such as Amazon SQS, Beanstalk, or Redis. However, any given queue connection may have multiple "queues" which may be thought of as different stacks or piles of queued jobs.

Note that each connection configuration example in the config.yml configuration file contains a queue attribute. This is the default queue that jobs will be dispatched to when they are sent to a given connection. In other words, if you dispatch a job without explicitly defining which queue it should be dispatched to, the job will be placed on the queue that is defined in the queue attribute of the connection configuration:

// This job is sent to the default queue...
$this->get('idb_queue')->push('service_id');

// This job is sent to the "emails" queue...
$this->get('idb_queue')->push('service_id', [], 'emails');

Some applications may not need to ever push jobs onto multiple queues, instead preferring to have one simple queue. However, pushing jobs to multiple queues can be especially useful for applications that wish to prioritize or segment how jobs are processed, since the QueueBundle queue worker allows you to specify which queues it should process by priority. For example, if you push jobs to a high queue, you may run a worker that gives them higher processing priority:

php console idb_queue:work --queue=high,default

Driver Prerequisites

Database

In order to use the database queue driver, you will need a run a following command which will generate necessary repository and entity to support the queue:

php console idb_queue:database

This will generate the necessary files in your cache/output folder, which you will need to move to appropriate location. Then define a service which definition will be shown during end of the command run.

Command assumes that you are running Doctrine (ORM or ODM) with annotation config. If you are using any other configuration format then you will have to make necessary adjustment in generated code.

Furthermore if you want to use relational database then you will need "doctrine/orm" if you want to use MongoDB then you will need "doctrine/mongodb-odm" and "doctrine/mongodb-odm-bundle".

If for any reason you need support for any other library besides Doctrine or any other database then please see here

Redis

In order to use the redis queue driver, you will need to have any service that provided Predis client instance. If you are using SNC RedisBundle then it will be snc_redis.default_client considering that you are using Predis as default Redis client.

Other Driver Prerequisites

The following dependencies are needed for the listed queue drivers:

Amazon SQS: aws/aws-sdk-php ~3.0
Beanstalkd: pda/pheanstalk ~3.0
Redis: predis/predis ~1.0

Creating Jobs

Generating Job Classes

Every Job is basically a service that implements IdeasBucket\QueueBundle\QueueableInterface. QueueBundle provides command for generating a job.

php console idb_queue:create_job

The generated class will be inside Job folder inside the bundle that you chose during command.

Job Structure

Job classes are very simple, it simply implements Queueable interface containing only a fire method which is called when the job is processed by the queue. To get started, let's take a look at an example job class. In this example, we'll send email using queue:

<?php

namespace AppBundle\Job;

use IdeasBucket\QueueBundle\QueueableInterface;
use IdeasBucket\QueueBundle\Job\JobsInterface;

class QueueMailer implements QueueableInterface
{
    /**
     * @var \Swift_Mailer
     */
    private $mailer;
    
    public function __construct(\Swift_Mailer $mailer)
    {
        $this->mailer = $mailer;
    }
    
    /**
     * @param JobsInterface $job
     * @param array         $data
     */
    public function fire(JobsInterface $job, array $data = null)
    {
        // Create a message
        //....
        $this->mailer->send($message);
        
        $jo

QueueBundle

Install / Use

README

QueueBundle

Introduction

Installation

Configuration

Connections vs. Queues

Driver Prerequisites

Database

Redis

Other Driver Prerequisites

Creating Jobs

Generating Job Classes

Job Structure