longhao / yii2-amqp
AMQP wrapper for Yii2 to publish and consume messages
Installs: 1
Dependents: 0
Suggesters: 0
Security: 0
Stars: 1
Watchers: 2
Forks: 0
Open Issues: 0
Type:project
Requires
- php: >=7.0
- php-amqplib/php-amqplib: 2.*
- yiisoft/yii2: ~2.0.18
This package is not auto-updated.
Last update: 2024-04-16 09:58:48 UTC
README
AMQP wrapper for Yii2 to publish and consume messages especially from RabbitMQ
Features
- Advanced queue configuration
- Add message to queues easily
- Listen queues with useful options
Installation
Composer
Add the following to your require part within the composer.json:
"longhao/yii2-amqp": "1.*"
$ php composer update
or
$ php composer require longhao/yii2-amqp
Integration
Yii2
Create a config folder in the root directory of your yii2 application and copy the content from vendor/longhao/yii2-amqp/config/amqp.php to common/config/params-local.php.
Adjust the properties to your needs.
return [ 'amqp' => [ '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']], 'exchange_properties' => [], 'timeout' => 0 ], ], ] ];
Register the Component in common/config/main.php:
/* |-------------------------------------------------------------------------- | Register Component |-------------------------------------------------------------------------- */ //... 'components' => [ 'Amqp' => [ 'class' => 'Longhao\Amqp\Amqp' ] ] //...
Register component is not necessary.
Yii2
Publishing a message
Push message with routing key
eg 1:
$amqp = new Amqp; $amqp->publish('routing-key', 'message');
eg 2:
Yii::$app->Amqp->publish('routing-key', 'message');
Push message with routing key and create queue
eg 1:
$amqp = new Amqp; $amqp->publish('routing-key', 'message' , ['queue' => 'queue-name']);
eg 2:
Yii::$app->Amqp->publish('routing-key', 'message' , ['queue' => 'queue-name']);
Push message with routing key and overwrite properties
eg 1:
$amqp = new Amqp; $amqp->publish('routing-key', 'message' , ['exchange' => 'amq.direct']);
eg 2:
Yii::$app->Amqp->publish('routing-key', 'message' , ['exchange' => 'amq.direct']);
Consuming messages
Consume messages, acknowledge and stop when no message is left
eg 1:
$amqp = new Amqp; $amqp->consume('queue-name', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); $resolver->stopWhenProcessed(); });
eg 2:
Yii::$ap->Amqp->consume('queue-name', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); $resolver->stopWhenProcessed(); });
Consume messages forever
eg 1:
$amqp = new Amqp; $amqp->consume('queue-name', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); });
eg 2:
Yii::$app->Amqp->consume('queue-name', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); });
Consume messages, with custom settings
eg 1:
$amqp = new Amqp; $amqp->consume('queue-name', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); }, [ 'timeout' => 2, 'vhost' => 'vhost3' ]);
eg 2:
Yii::$app->Amqp->consume('queue-name', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); }, [ 'timeout' => 2, 'vhost' => 'vhost3' ]);
Fanout example
Publishing a message
eg 1:
$amqp = new Amqp; $amqp->publish('', 'message' , [ 'exchange_type' => 'fanout', 'exchange' => 'amq.fanout', ]);
eg 2:
Yii::$app->Amqp->publish('', 'message' , [ 'exchange_type' => 'fanout', 'exchange' => 'amq.fanout', ]);
Consuming messages
eg 1:
$amqp = new Amqp; $amqp->consume('', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); }, [ 'exchange' => 'amq.fanout', 'exchange_type' => 'fanout', 'queue_force_declare' => true, 'queue_exclusive' => true, 'persistent' => true// required if you want to listen forever ]);
eg 2:
Yii::$app->Amqp->consume('', function ($message, $resolver) { var_dump($message->body); $resolver->acknowledge($message); }, [ 'exchange' => 'amq.fanout', 'exchange_type' => 'fanout', 'queue_force_declare' => true, 'queue_exclusive' => true, 'persistent' => true// required if you want to listen forever ]);
Credits
- Some concepts were used from https://github.com/mookofe/tail
License
This package is open-sourced software licensed under the MIT license