adtechpotok/enqueue-messenger-adapter-bundle
Symfony Messenger transport that uses Enqueue to send/receive messages via supported brokers (e.g., AMQP/RabbitMQ). Configure Messenger with enqueue:// DSNs, map queues/topics/exchanges, and consume with messenger:consume-messages.
This Symfony Messenger transport allows you to use Enqueue to send and receive your messages from all the supported brokers.
composer req enqueue/messenger-adapter
default Enqueue transport:# .env
# ...
###> enqueue/enqueue-bundle ###
ENQUEUE_DSN=amqp://guest:guest@localhost:5672/%2f
###< enqueue/enqueue-bundle ###
amqp) to use Enqueue's default transport:# config/packages/messenger.yaml
framework:
messenger:
transports:
amqp: enqueue://default
# config/packages/framework.yaml
framework:
messenger:
# ...
routing:
'App\Message\MyMessage': amqp
bin/console messenger:consume-messages amqp
In the transport DSN, you can add extra configuration. Here is the reference DSN (note that the values are just for the example):
enqueue://default
?queue[routingKey][name]=queue_name
&topic[name]=topic_name
&topic[type]=topic|fanout|direct
&deliveryDelay=1800
&delayStrategy=Adtechpotok\Bundle\EnqueueMessengerAdapterBundle\Transport\RabbitMq375DelayPluginDelayStrategy
&timeToLive=3600
&receiveTimeout=1000
&priority=1
&maximumPriority=255
&durability=1
# config/packages/messenger.yaml
framework:
messenger:
transports:
events: enqueue://default?queue[*][name]=events&topic[name]=events&topic[type]=topic
foo.events: enqueue://default?queue[foo][name]=foo.events&topic[name]=events&topic[type]=topic
bar.events: enqueue://default?queue[bar][name]=bar.events&topic[name]=events&topic[type]=topic
routing:
App\Message\EventsMessage: events
Foo\Message\EventsMessage: foo.events
Bar\Message\EventsMessage: bar.events
You can send a message on a specific topic using TransportConfiguration envelope item with your message:
use Enqueue\MessengerAdapter\EnvelopeItem\TransportConfiguration;
// ...
$this->bus->dispatch((new Envelope($message))->with(new TransportConfiguration(
['topic' => 'specific-topic']
)));
This package provide deduplication mechanism based on Redis optimistic locking (there is documentation https://redis.io/topics/transactions "Optimistic locking using check-and-set"). It uses two dependencies which you have to implement by yourself.
# config/packages/messenger.yaml
framework:
messenger:
buses:
messenger.produce:
middleware:
- Adtechpotok\Bundle\EnqueueMessengerAdapterBundle\Middleware\UuidItemSetterMiddleware
That interface will be used every time when LockBasedDeduplicationMiddleware will process a message. The method getUniqueId have to return a unique id each time when it will be called. LockBasedDeduplicationMiddleware will try to lock a message by this id. If it succeeds the message will be processed next otherwise you will get an exception.
Example:
class IdGenerator implements UniqueIdGetterInterface
{
/** @var string */
protected $id;
public function __construct()
{
$this->generateId();
}
/**
* @return string
*/
public function getUniqueId(): string
{
return $this->id;
}
public function generateId(): void
{
$this->id = uniqid('', true);
}
}
#config/services.yaml
services:
SystemBundle\Classes\IdGenerator:
system.middleware.service.locker:
class: Adtechpotok\Bundle\EnqueueMessengerAdapterBundle\Service\RedisLockService
arguments:
- '@REDIS_CLIENT'
- 'rabbit_mq_'
- 172800
- 'worker_id'
messenger.middleware.lock_based_deduplication:
class: Adtechpotok\Bundle\EnqueueMessengerAdapterBundle\Middleware\LockBasedDeduplicationMiddleware
arguments:
- '@system.middleware.service.locker'
- '@SystemBundle\Classes\IdGenerator'
Where @REDIS_CLIENT is your configured redis client.
common:
messenger:
buses:
messenger.consume:
middleware:
- messenger.middleware.lock_based_deduplication
How can I help you explore Laravel packages today?