swarrot/swarrot

A simple lib to consume RabbitMQ queues

v4.18.0 2024-02-20 20:50 UTC

Requires

• php: ^7.4 || ^8.0
• psr/log: ^1.0 || ^2.0 || ^3.0
• symfony/options-resolver: ^5.4 || ^6.0 || ^7.0

Requires (Dev)

• doctrine/dbal: ^3.0 || ^4.0
• doctrine/persistence: ^3.0
• friendsofphp/php-cs-fixer: ^3.27
• php-amqplib/php-amqplib: ^2.1 || ^3.0
• phpspec/prophecy: ^1.15
• phpspec/prophecy-phpunit: ^2.0
• phpstan/phpstan: ^1.10
• phpunit/phpunit: ^9.6 || ^10.0
• symfony/error-handler: ^5.4 || ^6.0 || ^7.0
• symfony/phpunit-bridge: ^5.4 || ^6.0 || ^7.0
• symfony/service-contracts: ^2.5 || ^3.0

Suggests

• pecl-amqp: *
• php-amqplib/php-amqplib: ^2.1 || ^3.0

Conflicts

• doctrine/persistence: <1.3

MIT

• Olivier Dolbeau

• queue
• AMQP
• worker
• swarrot

README

Swarrot is a PHP library to consume messages from any broker.

Installation

The recommended way to install Swarrot is through Composer. Require the swarrot/swarrot package:

$ composer require swarrot/swarrot

Usage

Basic usage

First, you need to create a message provider to retrieve messages from your broker. For example, with a PeclPackageMessageProvider (retrieves messages from an AMQP broker with the pecl amqp package:

use Swarrot\Broker\MessageProvider\PeclPackageMessageProvider;

// Create connection
$connection = new \AMQPConnection();
$connection->connect();
$channel = new \AMQPChannel($connection);
// Get the queue to consume
$queue = new \AMQPQueue($channel);
$queue->setName('global');

$messageProvider = new PeclPackageMessageProvider($queue);

Once it's done you need to create a Processor to process messages retrieved from the broker. This processor must implement Swarrot\Processor\ProcessorInterface. For example:

use Swarrot\Processor\ProcessorInterface;
use Swarrot\Broker\Message;

class Processor implements ProcessorInterface
{
    public function process(Message $message, array $options): bool
    {
        echo sprintf("Consume message #%d\n", $message->getId());

        return true; // Continue processing other messages
    }
}

You now have a Swarrot\Broker\MessageProviderInterface to retrieve messages and a Processor to process them. So, ask the Swarrot\Consumer to do its job :

use Swarrot\Consumer;

$consumer = new Consumer($messageProvider, $processor);
$consumer->consume();

Using a stack

Heavily inspired by stackphp/builder you can use Swarrot\Processor\Stack\Builder to stack your processors. Using the built in processors or by creating your own, you can extend the behavior of your base processor. In this example, your processor is decorated by 2 other processors. The ExceptionCatcherProcessor which decorates your own with a try/catch block and the MaxMessagesProcessor which stops your worker when some messages have been consumed.

use Swarrot\Processor\ProcessorInterface;
use Swarrot\Broker\Message;

class Processor implements ProcessorInterface
{
    public function process(Message $message, array $options): bool
    {
        echo sprintf("Consume message #%d\n", $message->getId());
        
        return true; // Continue processing other messages
    }
}

$stack = (new \Swarrot\Processor\Stack\Builder())
    ->push('Swarrot\Processor\MaxMessages\MaxMessagesProcessor', new Logger())
    ->push('Swarrot\Processor\ExceptionCatcher\ExceptionCatcherProcessor')
    ->push('Swarrot\Processor\Ack\AckProcessor', $messageProvider)
;

$processor = $stack->resolve(new Processor());

Here is an illustration to show you what happens when this order is used:

Processors

Official processors

• AckProcessor
• Doctrine related processors (thanks to Adrien Brault)
• ExceptionCatcherProcessor
• InsomniacProcessor (thanks to Adrien Brault)
• InstantRetryProcessor
• MaxExecutionTimeProcessor (thanks to Remy Lemeunier)
• MaxMessagesProcessor (thanks to Remy Lemeunier)
• MemoryLimitProcessor (thanks to Christophe Coevoet)
• RetryProcessor
• ServicesResetterProcessor (thanks to Pierrick Vignand)
• SignalHandlerProcessor
• XDeathMaxCountProcessor (thanks to Anthony Moutte)
• XDeathMaxLifetimeProcessor (thanks to Anthony Moutte)

Create your own processor

To create your own processor and be able to use it with the StackProcessor, you just need to implement ProcessorInterface and to take another ProcessorInterface as first argument in constructor.

Deprecated processors & message providers / publishers

In order to reduce swarrot/swarrot dependencies & ease the maintenance, some processors & message providers / publishers have been deprecated in 3.x version. They will be deleted in 4.0.

If you use those deprecated classes you could create your own repository to keep them or we could create a dedicated repository under the swarrot organisation if you're willing to help to maintain them.

Message providers / publishers

• SQS Message provider (in 3.5.0)
• Stomp message providers (in 3.6.0)
• Stomp message publishers (in 3.7.0)
• Interop message publishers & providers (in 3.7.0)

Processors

• SentryProcessor (in 3.5.0)
• RPC related processors (in 3.5.0)
• NewRelicProcessor (in 3.7.0)

Inspiration

• stackphp/builder

License

Swarrot is released under the MIT License. See the bundled LICENSE file for details.