RabbitMqBundle

RabbitMqBundle

About

The RabbitMqBundle incorporates messaging in your application via RabbitMQ using the php-amqplib library.

The bundle implements several messaging patterns as seen on the Thumper library. Therefore publishing messages to RabbitMQ from a Symfony controller is as easy as:

$msg = array('user_id' => 1235, 'image_path' => '/path/to/new/pic.png');
$this->get('old_sound_rabbit_mq.upload_picture_producer')->publish(serialize($msg));

Later when you want to consume 50 messages out of the upload_pictures queue, you just run on the CLI:

$ ./app/console rabbitmq:consumer -m 50 upload_picture

All the examples expect a running RabbitMQ server.

This bundle was presented at Symfony Live Paris 2011 conference. See the slides here.

Version 2

Due to the breaking changes happened caused by Symfony >=4.4, a new tag was released, making the bundle compatible with Symfony >=4.4.

Installation

For Symfony Framework >= 4.4

Require the bundle and its dependencies with composer:

$ composer require php-amqplib/rabbitmq-bundle

Register the bundle:

// app/AppKernel.php

public function registerBundles()
{
    $bundles = array(
        new OldSound\RabbitMqBundle\OldSoundRabbitMqBundle(),
    );
}

Enjoy !

For a console application that uses Symfony Console, Dependency Injection and Config components

If you have a console application used to run RabbitMQ consumers, you do not need Symfony HttpKernel and FrameworkBundle. From version 1.6, you can use the Dependency Injection component to load this bundle configuration and services, and then use the consumer command.

Require the bundle in your composer.json file:

{
    "require": {
        "php-amqplib/rabbitmq-bundle": "^2.0",
    }
}

Register the extension and the compiler pass:

use OldSound\RabbitMqBundle\DependencyInjection\OldSoundRabbitMqExtension;
use OldSound\RabbitMqBundle\DependencyInjection\Compiler\RegisterPartsPass;

// ...

$containerBuilder->registerExtension(new OldSoundRabbitMqExtension());
$containerBuilder->addCompilerPass(new RegisterPartsPass());

Warning - BC Breaking Changes

• Since 2012-06-04 Some default options for exchanges declared in the "producers" config section have changed to match the defaults of exchanges declared in the "consumers" section. The affected settings are:

  • durable was changed from false to true,
  • auto_delete was changed from true to false.

  Your configuration must be updated if you were relying on the previous default values.

• Since 2012-04-24 The ConsumerInterface::execute method signature has changed

• Since 2012-01-03 the consumers execute method gets the whole AMQP message object and not just the body. See the CHANGELOG file for more details.

Usage

Add the old_sound_rabbit_mq section in your configuration file:

…

Here we configure the connection service and the message endpoints that our application will have. In this example your service container will contain the service old_sound_rabbit_mq.upload_picture_producer and old_sound_rabbit_mq.upload_picture_consumer. The later expects that there's a service called upload_picture_service.

If you don't specify a connection for the client, the client will look for a connection with the same alias. So for our upload_picture the service container will look for an upload_picture connection.

If you need to add optional queue arguments, then your queue options can be something like this:

queue_options: {name: 'upload-picture', arguments: {'x-ha-policy': ['S', 'all']}}

another example with message TTL of 20 seconds:

queue_options: {name: 'upload-picture', arguments: {'x-message-ttl': ['I', 20000]}}

The argument value must be a list of datatype and value. Valid datatypes are:

• S - String
• I - Integer
• D - Decimal
• T - Timestamps
• F - Table
• A - Array
• t - Bool

Adapt the arguments according to your needs.

If you want to bind queue with specific routing keys you can declare it in producer or consumer config:

queue_options:
    name: "upload-picture"
    routing_keys:
      - 'android.#.upload'
      - 'iphone.upload'

Important notice - Lazy Connections

In a Symfony environment all services are fully bootstrapped for each request, from version >= 4.3 you can declare a service as lazy (Lazy Services). This bundle still doesn't support new Lazy Services feature but you can set lazy: true in your connection configuration to avoid unnecessary connections to your message broker in every request. It's extremely recommended to use lazy connections because performance reasons, nevertheless lazy option is disabled by default to avoid possible breaks in applications already using this bundle.

Important notice - Heartbeats

It's a good idea to set the read_write_timeout to 2x the heartbeat so your socket will be open. If you don't do this, or use a different multiplier, there's a risk the consumer socket will timeout.

Please bear in mind, that you can expect problems, if your tasks are generally running longer than the heartbeat period, to which there are no good solutions (link). Consider using either a big value for the heartbeat or leave the heartbeat disabled in favour of the tcp's keepalive (both on the client and server side) and the graceful_max_execution_timeout feature.

Multiple Hosts

You can provide multiple hosts for a connection. This will allow you to use RabbitMQ cluster with multiple nodes.

  old_sound_rabbit_mq:
      connections:
          default:
              hosts:
                - host: host1
                  port: 3672
                  user: user1
                  password: password1
                  vhost: vhost1
                - url: 'amqp://guest:password@localhost:5672/vhost'
              connection_timeout: 3
              read_write_timeout: 3

Pay attention that you can not specify

  connection_timeout 
  read_write_timeout
  use_socket
  ssl_context
  keepalive
  heartbeat
  connection_parameters_provider 

parameters to each host separately.

Dynamic Connection Parameters

Sometimes your connection information may need to be dynamic. Dynamic connection parameters allow you to supply or override parameters programmatically through a service.

e.g. In a scenario when the vhost parameter of the connection depends on the current tenant of your white-labeled application and you do not want (or can't) change it's configuration every time.

Define a service under connection_parameters_provider that implements the ConnectionParametersProviderInterface, and add it to the appropriate connections configuration.

connections:
    default:
        host:     'localhost'
        port:     5672
        user:     'guest'
        password: 'guest'
        vhost:    'foo' # to be dynamically overridden by `connection_parameters_provider`
        connection_parameters_provider: connection_parameters_provider_service

Example Implementation:

class ConnectionParametersProviderService implements ConnectionParametersProvider {
    ...
    public function getConnectionParameters() {
        return array('vhost' => $this->getVhost());
    }
    ...
}

In this case, the vhost parameter will be overridden by the output of getVhost().

Producers, Consumers, What?

In a messaging application, the process sending messages to the broker is called producer while the process receiving those messages is called consumer. In your application you will have several of them that you can list under their respective entries in the configuration.

Producer

A producer will be used to send messages to the server. In the AMQP Model, messages are sent to an exchange, this means that in the configuration for a producer you will have to specify the connection options along with the exchange options, which usually will be the name of the exchange and the type of it.

Now let's say that you want to process picture uploads in the background. After you move the picture to its final location, you will publish a message to server with the following information:

public function indexAction($name)
{
    $msg = array('user_id' => 1235, 'image_path' => '/path/to/new/pic.png');
    $this->get('old_sound_rabbit_mq.upload_picture_producer')->publish(serialize($msg));
}

As you can see, if in your configuration you have a producer called upload_picture, then in the service container you will have a service called old_sound_rabbit_mq.upload_picture_producer.

Besides the message itself, the OldSound\RabbitMqBundle\RabbitMq\Producer#publish() method also accepts an optional routing key parameter and an optional array of additional properties. The array of additional properties allows you to alter the properties with which an PhpAmqpLib\Message\AMQPMessage object gets constructed by default. This way, for example, you can change the application headers.

You can use setContentType and setDeliveryMode methods in order to set the message content type and the message delivery mode respectively, overriding any default set in the "producers" config section. If not overriden by either the "producers" configuration or an explicit call to these methods (as per the below example), the default values are text/plain for content type and 2 for delivery mode.

$this->get('old_sound_rabbit_mq.upload_picture_producer')->setContentType('application/json');

If you need to use a custom class for a producer (which should inherit from OldSound\RabbitMqBundle\RabbitMq\Producer), you can use the class option:

    ...
    producers:
        upload_picture:
            class: My\Custom\Producer
            connection: default
            exchange_options: {name: 'upload-picture', type: direct}
    ...

The next piece of the puzzle is to have a consumer that will take the message out of the queue and process it accordingly.

Producer Events

There are currently two events emitted by the producer.

BeforeProducerPublishMessageEvent

This event occurs immediately before publishing the message. This is a good hook to do any final logging, validation, etc. before actually sending the message. A sample implementation of a listener:

namespace App\EventListener;

use OldSound\RabbitMqBundle\Event\BeforeProducerPublishMessageEvent;
use Symfony\Component\EventDispatcher\Attribute\AsEventListener;

#[AsEventListener(event: BeforeProducerPublishMessageEvent::NAME)]
final class AMQPBeforePublishEventListener
{
    public function __invoke(BeforeProducerPublishMessageEvent $event): void
    {
        // Your code goes here
    }
}

AfterProducerPublishMessageEvent

This event occurs immediately after publishing the message. This is a good hook to do any confirmation logging, commits, etc. after actually sending the message. A sample implementation of a listener:

namespace App\EventListener;

use OldSound\RabbitMqBundle\Event\AfterProducerPublishMessageEvent;
use Symfony\Component\EventDispatcher\Attribute\AsEventListener;

#[AsEventListener(event: AfterProducerPublishMessageEvent::NAME)]
final class AMQPBeforePublishEventListener
{
    public function __invoke(AfterProducerPublishMessageEvent $event): void
    {
        // Your code goes here
    }
}

Consumers

A consumer will connect to the server and start a __l