A lib to consume message from any Broker
Swarrot is a PHP library to consume messages from any broker.
The recommended way to install Swarrot is through
Composer. Require the swarrot/swarrot
package:
$ composer require swarrot/swarrot
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();
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:
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.
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.
Swarrot is released under the MIT License. See the bundled LICENSE file for details.