AMQP wrapper for Laravel and Lumen to publish and consume messages especially from RabbitMQ
- Advanced queue configuration
- Add message to queues easily
- Listen queues with useful options
Add the following to your require part within the composer.json:
"bschmitt/laravel-amqp": "2.*" (Laravel >= 5.5)
"bschmitt/laravel-amqp": "1.*" (Laravel < 5.5)
$ php composer update
or
$ php composer require bschmitt/laravel-amqp
Create a config folder in the root directory of your Lumen application and copy the content from vendor/bschmitt/laravel-amqp/config/amqp.php to config/amqp.php.
Adjust the properties to your needs.
return [
'use' => 'production',
'properties' => [
'production' => [
'host' => 'localhost',
'port' => 5672,
'username' => 'username',
'password' => 'password',
'vhost' => '/',
'exchange' => 'amq.topic',
'exchange_type' => 'topic',
'consumer_tag' => 'consumer',
'ssl_options' => [], // See https://secure.php.net/manual/en/context.ssl.php
'connect_options' => [], // See https://github.com/php-amqplib/php-amqplib/blob/master/PhpAmqpLib/Connection/AMQPSSLConnection.php
'queue_properties' => ['x-ha-policy' => ['S', 'all'], 'x-max-priority' => ['I', 5]],
'exchange_properties' => [],
'timeout' => 0
],
],
];
Register the Lumen Service Provider in bootstrap/app.php:
/*
|--------------------------------------------------------------------------
| Register Service Providers
|--------------------------------------------------------------------------
*/
//...
$app->configure('amqp');
$app->register(Bschmitt\Amqp\LumenServiceProvider::class);
//...
Add Facade Support for Lumen 5.2+
//...
$app->withFacades(true, [
'Bschmitt\Amqp\Facades\Amqp' => 'Amqp',
]);
//...
Open config/app.php and add the service provider and alias:
'Bschmitt\Amqp\AmqpServiceProvider',
'Amqp' => 'Bschmitt\Amqp\Facades\Amqp',
Amqp::publish('routing-key', 'message');
Amqp::publish('routing-key', 'message' , ['queue' => 'queue-name']);
Amqp::publish('routing-key', 'message' , ['exchange' => 'amq.direct']);
Amqp::consume('queue-name', function ($message, $resolver) {
var_dump($message->body);
$resolver->acknowledge($message);
$resolver->stopWhenProcessed();
});
Amqp::consume('queue-name', function ($message, $resolver) {
var_dump($message->body);
$resolver->acknowledge($message);
});
Amqp::consume('queue-name', function ($message, $resolver) {
var_dump($message->body);
$resolver->acknowledge($message);
}, [
'timeout' => 2,
'vhost' => 'vhost3'
]);
\Amqp::publish('', 'message' , [
'exchange_type' => 'fanout',
'exchange' => 'amq.fanout',
]);
\Amqp::consume('', function ($message, $resolver) {
var_dump($message->body);
$resolver->acknowledge($message);
}, [
'routing' => '',
'exchange' => 'amq.fanout',
'exchange_type' => 'fanout',
'queue_force_declare' => true,
'queue_exclusive' => true,
'persistent' => true // required if you want to listen forever
]);
Amqp::publish('routing-key', 'I will be processed second.' , ['queue' => 'queue-name', 'priority' => 0]);
Amqp::publish('routing-key', 'I will be processed first.' , ['queue' => 'queue-name', 'priority' => 1]);
- Some concepts were used from https://github.com/mookofe/tail
This package is open-sourced software licensed under the MIT license