Swarrot
Swarrot is PHP library to consume messages from any broker.
Installation
The recommended way to install Swarrot is through
Composer. Require the swarrot/swarrot
package
into your composer.json
file:
{
"require": {
"swarrot/swarrot": "@stable"
}
}
Protip: you should browse the
swarrot/swarrot
page to choose a stable version to use, avoid the @stable
meta constraint.
Usage
Basic usage
First, you need to create a message provider to retrieve message from your broker. For example, with A PeclPackageMessageProvider (retrieve message 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) {
echo sprintf("Consume message #%d\n", $message->getId());
}
}
You now have a Swarrot\Broker\MessageProviderInterface
to retrieve messages
and a Processor to process them. So, ask the Swarrot\Consumer
to do it's 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 others one. The
ExceptionCatcherProcessor
which decorate your own with a try/catch block and the
MaxMessagesProcessor
which stop 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) {
echo sprintf("Consume message #%d\n", $message->getId());
}
}
$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 append 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)
- NewRelicProcessor (thanks to Adrien Brault)
- RetryProcessor
- RPC related processors (thanks to Baptiste Clavié)
- SignalHandlerProcessor
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.
Inspiration
License
Swarrot is released under the MIT License. See the bundled LICENSE file for details.