evaneos / burrow
AMQP Messaging Event component
Installs: 98 554
Dependents: 4
Suggesters: 1
Security: 0
Stars: 3
Watchers: 25
Forks: 1
Open Issues: 3
Requires
- php: >=5.5.9
- ext-pcntl: *
- beberlei/assert: ^2.6
- beberlei/metrics: ^2.8
- evaneos/daemon: ^2.0
- league/event: ^2.2
- php-amqplib/php-amqplib: ^2.6
- psr/log: ~1.0
- symfony/console: ^2.8|^3.0|^4.0
Requires (Dev)
- fzaninotto/faker: ^1.6
- mockery/mockery: ^0.9.4
- monolog/monolog: ~1.13
- phpunit/phpunit: ~4.0|~5.0
- squizlabs/php_codesniffer: ~2.0
- symfony/process: ^3.2
This package is auto-updated.
Last update: 2024-11-05 23:49:32 UTC
README
Evaneos AMQP library able to use both php-amqplib and pecl amqp C librairy
Installation
composer require evaneos/burrow
Usage
See examples directory for more details
To test it, you may use a rabbitmq container, this one feets perfectly
docker run -d -p 5672:5672 rabbitmq
Declare an exchange and bind a queue with RabbitMQ
$admin = DriverFactory::getDriver([ 'host' => 'localhost', 'port' => '5672', 'user' => 'guest', 'pwd' => 'guest' ]); $admin->declareExchange('exchange'); $admin->declareAndBindQueue('exchange', 'my_queue');
Asynchronous management
Dispatching an async message with RabbitMQ
$driver = DriverFactory::getDriver([ 'host' => 'localhost', 'port' => '5672', 'user' => 'guest', 'pwd' => 'guest' ]); $publisher = new AsyncPublisher($driver, 'exchange'); $publisher->publish('message', 'routing_key', [ 'meta' => 'data' ]);
Write a daemon to consume async messages from RabbitMQ
$driver = DriverFactory::getDriver([ 'host' => 'default', 'port' => '5672', 'user' => 'guest', 'pwd' => 'guest' ]); $handlerBuilder = new HandlerBuilder($driver); $handler = $handlerBuilder->async()->build(new EchoConsumer()); $daemon = new QueueHandlingDaemon($driver, $handler, 'test'); $worker = new Worker($daemon); $worker->run();
In the command-line, launch both scripts from a different terminal, the message 'my_message', should be displayed in the worker terminal.
Synchronous management
Dispatching an async message with RabbitMQ
$driver = DriverFactory::getDriver([ 'host' => 'default', 'port' => '5672', 'user' => 'guest', 'pwd' => 'guest' ]); $publisher = new SyncPublisher($driver, 'xchange'); $publisher->publish('my_message', 'routing_key', [ 'meta' => 'data' ]);
Write a daemon to consume async messages from RabbitMQ
$driver = DriverFactory::getDriver([ 'host' => 'default', 'port' => '5672', 'user' => 'guest', 'pwd' => 'guest' ]); $handlerBuilder = new HandlerBuilder($driver); $handler = $handlerBuilder->sync()->build(new ReturnConsumer()); $daemon = new QueueHandlingDaemon($driver, $handler, 'test'); $worker = new Worker($daemon); $worker->run();
In the command-line, launch both scripts from a different terminal, the message 'my_message', should be displayed in the publisher terminal.
Events
You can add your emitter to subscribe events:
$emitter = new League\Event\Emitter(); $emitter->addListener(new MyListener()); new QueueHandlingDaemon([..], $emitter);
Metrics
Based on events, you can subscribe a built-in metric publisher which will send this metrics:
daemon.started
(increment)daemon.stopped
(increment)daemon.message_received
(increment)daemon.message_consumed
(increment)daemon.message_processing_time
(timing)
There is an implementation for StatsD and DogStatsD.
$config = ['host' => 'host', 'port' => 'port']; // StatsD $metricService = MetricServiceFactory::create('statsd', $config); // DogStatsD $tags = ['service' => 'myService']; // This tags will be sent with all the metrics $metricService = MetricServiceFactory::create('dogstats', $config, $tags); $emitter = new League\Event\Emitter(); $emitter->useListenerProvider(new SendMetricListenerProvider($metricService)); new QueueHandlingDaemon([..], $emitter);
Examples
All these examples are also available in the example
directory.
Handlers
You can now use handlers to modify the consumption behaviour. For retro-compatibility reasons, a
ContinueOnFailureHandler
has been created to reproduce the previous default behaviour. Please, do not use it anymore
as it is quite dangerous to allow the worker to continue when receiving an error.
To ease the use of the handlers, please build the handler stack using HandlerBuilder
.