README

Getting Started

This package is a part of Psyllium library. It provides a simple way to use RabbitMQ in your application. Allows you to define listeners and message producers in the queue system.

Install

To install the package you will need to be using Composer in your project. To install it please see the docs.

composer require psyllium/amqp

Enable the Bundle

Enable it by adding it to the list of registered bundles in the Kernel.php file of your project:

config/bundles.php

return [
    Psyllium\Amqp\PsylliumAmqp::class => ['all' => true]
];

config/packages/psyllium_amqp.yaml

psyllium_amqp:
  rabbitmq:
    hostname: '%hostname%'
    port: '%port%'
    user: '%user%'
    pass: '%password%'
    prefetch_count: 1

prefetch_count is optional and defaults to 1: the broker delivers one message at a time and waits for its acknowledgement before sending the next one. Raise it only for consumers whose handlers are idempotent, because every unacknowledged message in flight is redelivered when the consumer dies. Values below 1 are rejected when the connection is opened, so the option can safely be bound to an environment variable such as '%env(int:RABBITMQ_PREFETCH_COUNT)%'.

Configure

In this way, you can create producers and listeners who will use the connection to the queuing system generated by the bundle.

config/services.yaml

psyllium.rabbitmq.listener:
  class: 'Psyllium\Amqp\Infrastructure\RabbitMQ\Listener'
  factory: ['@Psyllium\Amqp\Infrastructure\RabbitMQ\ListenerFactory', 'create']
  arguments:
    - '%queue_name%'

config/services.yaml

psyllium.rabbitmq.producer:
  class: 'Psyllium\Amqp\Infrastructure\RabbitMQ\Producer'
  factory: [ '@Psyllium\Amqp\Infrastructure\RabbitMQ\ProducerFactory', 'create' ]
  arguments:
    - '%exchange_name%'

Both configurations ( producers and consumers ) allow the definition of middlewares that can be used to transform messages as sequential steps performed in the order defined before the final delivery / publication of the message. Example:

psyllium.rabbitmq.listener:
  class: 'Psyllium\Amqp\Infrastructure\RabbitMQ\Listener'
  factory: ['@Psyllium\Amqp\Infrastructure\RabbitMQ\ListenerFactory', 'create']
  arguments:
    - '%queue_name%'
    - ['@Psyllium\Amqp\Domain\Middleware\GzUncompress', '@Psyllium\Amqp\Domain\Middleware\Timestamp']

How to use in code?

Listener:

use Psyllium\Amqp\Domain\ReadModel\ReadInterface;

$this->listener->listen(
    function (ReadInterface $message): void {
        $message->getBody()
        // do something with message
        $message->ack();
    }
);

Producer:

Definde class that implements EventInterface

use Psyllium\Amqp\Domain\EventModel\EventInterface;

class Example implements EventInterface

Send message to queue

$this->producer->produce(
    new Example()
);

Middlewares

Definde class that implements MiddlewareInterface

use Psyllium\Amqp\Domain\Middleware\MiddlewareInterface;

class Example implements MiddlewareInterface

The system also has 3 pre-defined middlewares to compress, decompress and timestamp messages.

@Psyllium\Amqp\Domain\Middleware\GzUncompress

Decompression of messages using the gzuncompress function.

@Psyllium\Amqp\Domain\Middleware\GzCompress

Compression of messages using the gzcompress function. Allows the specification of the compression level ( from -1 to 9, default: 9 ). Example:

psyllium.rabbitmq.middleware.gzcompress:
  class: 'Psyllium\Amqp\Domain\Middleware\GzCompress'
  arguments:
    $compressionLevel: 9

@Psyllium\Amqp\Domain\Middleware\Timestamp

Add a timestamp in the message header. Requires to define the name of the parameter. Example:

psyllium.rabbitmq.middleware.timestamp:
  class: 'Psyllium\Amqp\Domain\Middleware\Timestamp'
  arguments:
    $key: 'default-name'

psyllium / amqp

Package info

Security

Statistics