Skip to content
Uses Enqueue with Symfony's Messenger component.
Branch: master
Clone or download
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
Bundle Add Symfony 4.3 combatibility May 16, 2019
EnvelopeItem Fixing serialization problems with ObjectNormalizer + more helpers Mar 12, 2019
Exception Add AMQP topic exchange handling Nov 11, 2018
Tests Add Symfony 4.3 combatibility May 16, 2019
.gitignore WIP: Tests + small fixes (#9) Mar 27, 2018
.php_cs.dist WIP: Tests + small fixes (#9) Mar 27, 2018
.travis.yml Travis: Test on different composer stability. Nov 30, 2018
AmqpContextManager.php Add AMQP topic exchange handling Nov 11, 2018
ContextManager.php Fixed merge conflicts and other small things Oct 16, 2018
LICENSE Add the missing LICENSE Apr 12, 2018
MessageBusProcessor.php Fixed tests and compatibility issues with latest composer packages Oct 27, 2018
QueueInteropTransport.php Add Symfony 4.3 combatibility May 16, 2019
QueueInteropTransportFactory.php Add Symfony 4.3 combatibility May 16, 2019
README.md enrich documentation related to Kafka Jun 3, 2019
composer.json Rename package to `sroze/messenger-enqueue-transport` Jun 2, 2019
phpunit.xml.dist

README.md

Enqueue's transport for Symfony Messenger component

This Symfony Messenger transport allows you to use Enqueue to send and receive your messages from all the supported brokers.

Usage

  1. Install the transport
composer req sroze/messenger-enqueue-transport
  1. 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 ###
  1. Configure Messenger's transport (that we will name amqp) to use Enqueue's default transport:
# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            amqp: enqueue://default
  1. Route the messages that have to go through the message queue:
# config/packages/framework.yaml
framework:
    messenger:
        # ...

        routing:
            'App\Message\MyMessage': amqp
  1. Consume!
bin/console messenger:consume-messages 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),
    ]
])))
You can’t perform that action at this time.