sroze / messenger-enqueue-transport
Enqueue adapter for Symfony Messenger component
Installs: 7 624 050
Dependents: 2
Suggesters: 0
Security: 0
Stars: 191
Watchers: 17
Forks: 55
Open Issues: 40
Type:symfony-bundle
Requires
- php: ^8.1
- enqueue/amqp-tools: ^0.10
- enqueue/enqueue-bundle: ^0.10
- symfony/messenger: ^5.4 || ^6.3 || ^7.0
- symfony/options-resolver: ^5.4 || ^6.3 || ^7.0
Requires (Dev)
- enqueue/snsqs: ^0.10.11
- phpspec/prophecy: ^1.15
- phpspec/prophecy-phpunit: ^2.0
- phpunit/phpunit: ^10.0
- symfony/yaml: ^5.4 || ^6.3 || ^7.0
Replaces
- enqueue/messenger-adapter: >0.2.2
README
This Symfony Messenger transport allows you to use Enqueue to send and receive your messages from all the supported brokers.
Usage
- Install the transport
composer req sroze/messenger-enqueue-transport
- Configure the Enqueue bundle as you would normaly do (see Enqueue's Bundle documentation). If you are using the recipes, you should
just have to configure the environment variables to configure the
default
Enqueue transport:
# .env # ... ###> enqueue/enqueue-bundle ### ENQUEUE_DSN=amqp://guest:guest@localhost:5672/%2f ###< enqueue/enqueue-bundle ###
- Configure Messenger's transport (that we will name
amqp
) to use Enqueue'sdefault
transport:
# config/packages/messenger.yaml framework: messenger: transports: amqp: enqueue://default
- Route the messages that have to go through the message queue:
# config/packages/messenger.yaml framework: messenger: # ... routing: 'App\Message\MyMessage': amqp
- Consume!
bin/console messenger:consume amqp
Advanced usage
Configure the queue(s) and exchange(s)
In the transport DSN, you can add extra configuration. Here is the common reference DSN (note that the values are just for the example):
enqueue://default
?queue[name]=queue_name
&topic[name]=topic_name
&deliveryDelay=1800
&delayStrategy=Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy
&timeToLive=3600
&receiveTimeout=1000
&priority=1
Setting Custom Configuration on your Message
Each Enqueue transport (e.g. amqp, redis, etc) has its own message object
that can normally be configured by calling setter methods (e.g.
$message->setDeliveryDelay(5000)
). But in Messenger, you don't have access
to these objects directly. Instead, you can set them indirectly via
the TransportConfiguration
stamp:
use Symfony\Component\Messenger\Envelope; use Enqueue\MessengerAdapter\EnvelopeItem\TransportConfiguration; // ... // create your message like normal $message = // ... $transportConfig = (new TransportConfiguration()) // commmon options have a convenient method ->setDeliveryDelay(5000) // other transport-specific options are set via metadata // example custom option for AmqpMessage // each "metadata" will map to a setter on your message // will result in setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT) // being called ->addMetadata('deliveryMode', AmqpMessage::DELIVERY_MODE_PERSISTENT) ; $bus->dispatch((new Envelope($message))->with($transportConfig));
Send a message on a specific topic
You can send a message on a specific topic using TransportConfiguration
envelope item with your message:
use Symfony\Component\Messenger\Envelope; use Enqueue\MessengerAdapter\EnvelopeItem\TransportConfiguration; // ... $transportConfig = (new TransportConfiguration()) ->setTopic('specific-topic') ; $bus->dispatch((new Envelope($message))->with($transportConfig));
Use AMQP topic exchange
See https://www.rabbitmq.com/tutorials/tutorial-five-php.html
You can use specific topic and queue options to configure your AMQP exchange in topic
mode and bind it:
enqueue://default
?queue[name]=queue_name
&queue[bindingKey]=foo.#
&topic[name]=topic_name
&topic[type]=topic
&deliveryDelay=1800
&delayStrategy=Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy
&timeToLive=3600
&receiveTimeout=1000
&priority=1
Here is the way to send a message with a routing key matching this consumer:
$bus->dispatch((new Envelope($message))->with(new TransportConfiguration([ 'topic' => 'topic_name', 'metadata' => [ 'routingKey' => 'foo.bar' ] ])));
Configure custom Kafka message
Here is the way to send a message with with some custom options:
$this->bus->dispatch((new Envelope($message))->with(new TransportConfiguration([ 'topic' => 'test_topic_name', 'metadata' => [ 'key' => 'foo.bar', 'partition' => 0, 'timestamp' => (new \DateTimeImmutable())->getTimestamp(), 'messageId' => uniqid('kafka_', true), ] ])))