Messenger: Sync & Queued Message Handling

Messenger: Sync & Queued Message Handling

Messenger provides a message bus with the ability to send messages and thenhandle them immediately in your application or send them through transports(e.g. queues) to be handled later. To learn more deeply about it, read theMessenger component docs.

Installation

In applications using Symfony Flex, run this command toinstall messenger:

$ composer require messenger

Creating a Message & Handler

Messenger centers around two different classes that you'll create: (1) a messageclass that holds data and (2) a handler(s) class that will be called when thatmessage is dispatched. The handler class will read the message class and performsome task.

There are no specific requirements for a message class, except that it can beserialized:

// src/Message/SmsNotification.php
namespace App\Message;
 
class SmsNotification
{
    private $content;
 
    public function __construct(string $content)
    {
        $this->content = $content;
    }
 
    public function getContent(): string
    {
        return $this->content;
    }
}

A message handler is a PHP callable, the easiest way to create it is to create a class that implementsMessageHandlerInterface and has an __invoke() method that'stype-hinted with the message class (or a message interface):

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;
 
use App\Message\SmsNotification;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
 
class SmsNotificationHandler implements MessageHandlerInterface
{
    public function __invoke(SmsNotification $message)
    {
        // ... do some work - like sending an SMS message!
    }
}

Thanks to autoconfiguration and the SmsNotificationtype-hint, Symfony knows that this handler should be called when an SmsNotificationmessage is dispatched. Most of the time, this is all you need to do. But you canalso manually configure message handlers. Tosee all the configured handlers, run:

$ php bin/console debug:messenger

Dispatching the Message

You're ready! To dispatch the message (and call the handler), inject themessage_bus service (via the MessageBusInterface), like in a controller:

// src/Controller/DefaultController.php
namespace App\Controller;
 
use App\Message\SmsNotification;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\Messenger\MessageBusInterface;
 
class DefaultController extends AbstractController
{
    public function index(MessageBusInterface $bus)
    {
        // will cause the SmsNotificationHandler to be called
        $bus->dispatch(new SmsNotification('Look! I created a message!'));
 
        // or use the shortcut
        $this->dispatchMessage(new SmsNotification('Look! I created a message!'));
 
        // ...
    }
}

Transports: Async/Queued Messages

By default, messages are handled as soon as they are dispatched. If you wantto handle a message asynchronously, you can configure a transport. A transportis capable of sending messages (e.g. to a queueing system) and thenreceiving them via a worker. Messenger supportsmultiple transports.

Note

If you want to use a transport that's not supported, check out theEnqueue's transport, which supports things like Kafka, Amazon SQS andGoogle Pub/Sub.

A transport is registered using a "DSN". Thanks to Messenger's Flex recipe, your.env file already has a few examples.

# MESSENGER_TRANSPORT_DSN=amqp://guest:[email protected]:5672/%2f/messages
# MESSENGER_TRANSPORT_DSN=doctrine://default
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages

Uncomment whichever transport you want (or set it in .env.local). SeeTransport Configuration for more details.

Next, in config/packages/messenger.yaml, let's define a transport called asyncthat uses this configuration:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async: "%env(MESSENGER_TRANSPORT_DSN)%"
 
            # or expanded to configure more options
            #async:
            #    dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
            #    options: []

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
 
    <framework:config>
        <framework:messenger>
            <framework:transport name="async">%env(MESSENGER_TRANSPORT_DSN)%</framework:transport>
 
            <!-- or expanded to configure more options -->
            <framework:transport name="async"
                dsn="%env(MESSENGER_TRANSPORT_DSN)%"
            >
                <option key="...">...</option>
            </framework:transport>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'transports' => [
            'async' => '%env(MESSENGER_TRANSPORT_DSN)%',
 
            // or expanded to configure more options
            'async' => [
               'dsn' => '%env(MESSENGER_TRANSPORT_DSN)%',
               'options' => []
            ],
        ],
    ],
]);

Routing Messages to a Transport

Now that you have a transport configured, instead of handling a message immediately,you can configure them to be sent to a transport:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async: "%env(MESSENGER_TRANSPORT_DSN)%"
 
        routing:
            # async is whatever name you gave your transport above
            'App\Message\SmsNotification':  async

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
 
    <framework:config>
        <framework:messenger>
            <framework:routing message-class="App\Message\SmsNotification">
                <!-- async is whatever name you gave your transport above -->
                <framework:sender service="async"/>
            </framework:routing>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'routing' => [
            // async is whatever name you gave your transport above
            'App\Message\SmsNotification' => 'async',
        ],
    ],
]);

Thanks to this, the App\Message\SmsNotification will be sent to the asynctransport and its handler(s) will not be called immediately. Any messages notmatched under routing will still be handled immediately.

You can also route classes by their parent class or interface. Or send messagesto multiple transport:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        routing:
            # route all messages that extend this example base class or interface
            'App\Message\AbstractAsyncMessage': async
            'App\Message\AsyncMessageInterface': async
 
            'My\Message\ToBeSentToTwoSenders': [async, audit]

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
 
    <framework:config>
        <framework:messenger>
            <!-- route all messages that extend this example base class or interface -->
            <framework:routing message-class="App\Message\AbstractAsyncMessage">
                <framework:sender service="async"/>
            </framework:routing>
            <framework:routing message-class="App\Message\AsyncMessageInterface">
                <framework:sender service="async"/>
            </framework:routing>
            <framework:routing message-class="My\Message\ToBeSentToTwoSenders">
                <framework:sender service="async"/>
                <framework:sender service="audit"/>
            </framework:routing>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'routing' => [
            // route all messages that extend this example base class or interface
            'App\Message\AbstractAsyncMessage' => 'async',
            'App\Message\AsyncMessageInterface' => 'async',
            'My\Message\ToBeSentToTwoSenders' => ['async', 'audit'],
        ],
    ],
]);

Doctrine Entities in Messages

If you need to pass a Doctrine entity in a message, it's better to pass the entity'sprimary key (or whatever relevant information the handler actually needs, like email,etc) instead of the object:

class NewUserWelcomeEmail
{
    private $userId;
 
    public function __construct(int $userId)
    {
        $this->userId = $userId;
    }
 
    public function getUserId(): int
    {
        return $this->userId;
    }
}

Then, in your handler, you can query for a fresh object:

// src/MessageHandler/NewUserWelcomeEmailHandler.php
namespace App\MessageHandler;
 
use App\Message\NewUserWelcomeEmail;
use App\Repository\UserRepository;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
 
class NewUserWelcomeEmailHandler implements MessageHandlerInterface
{
    private $userRepository;
 
    public function __construct(UserRepository $userRepository)
    {
        $this->userRepository = $userRepository;
    }
 
    public function __invoke(NewUserWelcomeEmail $welcomeEmail)
    {
        $user = $this->userRepository->find($welcomeEmail->getUserId());
 
        // ... send an email!
    }
}

This guarantees the entity contains fresh data.

Handling Messages Synchronously

If a message doesn't match any routing rules, it won'tbe sent to any transport and will be handled immediately. In some cases (likewhen binding handlers to different transports),it's easier or more flexible to handle this explicitly: by creating a synctransport and "sending" messages there to be handled immediately:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            # ... other transports
 
            sync: 'sync://'
 
        routing:
            App\Message\SmsNotification: sync

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
 
    <framework:config>
        <framework:messenger>
            <!-- ... other transports -->
 
            <framework:transport name="sync" dsn="sync://"/>
 
            <framework:routing message-class="App\Message\SmsNotification">
                <framework:sender service="sync"/>
            </framework:routing>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'transports' => [
            // ... other transports
 
            'sync' => 'sync://',
        ],
        'routing' => [
            'App\Message\SmsNotification' => 'sync',
        ],
    ],
]);

Creating your Own Transport

You can also create your own transport if you need to send or receive messagesfrom something that is not supported. See How to Create Your own Messenger Transport.

Consuming Messages (Running the Worker)

Once your messages have been routed, in most cases, you'll need to "consume" them.You can do this with the messenger:consume command:

$ php bin/console messenger:consume async
 
# use -vv to see details about what's happening
$ php bin/console messenger:consume async -vv

New in version 4.3: The messenger:consume command was renamed in Symfony 4.3 (previously itwas called messenger:consume-messages).

The first argument is the receiver's name (or service id if you routed to acustom service). By default, the command will run forever: looking for new messageson your transport and handling them. This command is called your "worker".

Deploying to Production

On production, there are a few important things to think about:

Use Supervisor to keep your worker(s) running
You'll want one or more "workers" running at all times. To do that, use aprocess control system like Supervisor.
Don't Let Workers Run Forever
Some services (like Doctrine's EntityManager) will consume more memoryover time. So, instead of allowing your worker to run forever, use a flaglike messenger:consume —limit=10 to tell your worker to only handle 10messages before exiting (then Supervisor will create a new process). Thereare also other options like —memory-limit=128M and —time-limit=3600.
Restart Workers on Deploy
Each time you deploy, you'll need to restart all your worker processes sothat they see the newly deployed code. To do this, run messenger:stop-workerson deploy. This will signal to each worker that it should finish the messageit's currently handling and shut down gracefully. Then, Supervisor will createnew worker processes. The command uses the appcache internally - so make sure this is configured to use an adapter you like.

Prioritized Transports

Sometimes certain types of messages should have a higher priority and be handledbefore others. To make this possible, you can create multiple transports and routedifferent messages to them. For example:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    # queue_name is specific to the doctrine transport
                    queue_name: high
 
                    # for amqp send to a separate exchange then queue
                    #exchange:
                    #    name: high
                    #queues:
                    #    messages_high: ~
                    # or redis try "group"
            async_priority_low:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: low
 
        routing:
            'App\Message\SmsNotification':  async_priority_low
            'App\Message\NewUserWelcomeEmail':  async_priority_high

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
 
    <framework:config>
        <framework:messenger>
            <framework:transport name="async_priority_high" dsn="%env(MESSENGER_TRANSPORT_DSN)%">
                <option key="queue_name">high</option>
            </framework:transport>
            <framework:transport name="async_priority_low" dsn="%env(MESSENGER_TRANSPORT_DSN)%">
                <option key="queue_name">low</option>
            </framework:transport>
 
            <framework:routing message-class="App\Message\SmsNotification">
                <framework:sender service="async_priority_low"/>
            </framework:routing>
            <framework:routing message-class="App\Message\NewUserWelcomeEmail">
                <framework:sender service="async_priority_high"/>
            </framework:routing>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'transports' => [
            'async_priority_high' => [
                'dsn' => '%env(MESSENGER_TRANSPORT_DSN)%',
                'options' => [
                    'queue_name' => 'high',
                ],
            ],
            'async_priority_low' => [
                'dsn' => '%env(MESSENGER_TRANSPORT_DSN)%',
                'options' => [
                    'queue_name' => 'low',
                ],
            ],
        ],
        'routing' => [
            'App\Message\SmsNotification' => 'async_priority_low',
            'App\Message\NewUserWelcomeEmail' => 'async_priority_high',
        ],
    ],
]);

You can then run individual workers for each transport or instruct one workerto handle messages in a priority order:

$ php bin/console messenger:consume async_priority_high async_priority_low

The worker will always first look for messages waiting on asyncpriority_high. Ifthere are none, _then it will consume messages from async_priority_low.

Supervisor Configuration

Supervisor is a great tool to guarantee that your worker process(es) isalways running (even if it closes due to failure, hitting a message limitor thanks to messenger:stop-workers). You can install it on Ubuntu, forexample, via:

$ sudo apt-get install supervisor

Supervisor configuration files typically live in a /etc/supervisor/conf.ddirectory. For example, you can create a new messenger-worker.conf filethere to make sure that 2 instances of messenger:consume are running at alltimes:

;/etc/supervisor/conf.d/messenger-worker.conf
[program:messenger-consume]
command=php /path/to/your/app/bin/console messenger:consume async --time-limit=3600
user=ubuntu
numprocs=2
autostart=true
autorestart=true
process_name=%(program_name)s_%(process_num)02d

Change the async argument to use the name of your transport (or transports)and user to the Unix user on your server. Next, tell Supervisor to read yourconfig and start your workers:

$ sudo supervisorctl reread
 
$ sudo supervisorctl update
 
$ sudo supervisorctl start messenger-consume:*

See the Supervisor docs for more details.

Retries & Failures

If an exception is thrown while consuming a message from a transport it willautomatically be re-sent to the transport to be tried again. By default, a messagewill be retried 3 times before being discarded orsent to the failure transport. Each retrywill also be delayed, in case the failure was due to a temporary issue. All ofthis is configurable for each transport:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
 
                # default configuration
                retry_strategy:
                    max_retries: 3
                    # milliseconds delay
                    delay: 1000
                    # causes the delay to be higher before each retry
                    # e.g. 1 second delay, 2 seconds, 4 seconds
                    multiplier: 2
                    max_delay: 0
                    # override all of this with a service that
                    # implements Symfony\Component\Messenger\Retry\RetryStrategyInterface
                    # service: null

Avoiding Retrying

Sometimes handling a message might fail in a way that you know is permanentand should not be retried. If you throwUnrecoverableMessageHandlingException,the message will not be retried.

Saving & Retrying Failed Messages

If a message fails it is retried multiple times (max_retries) and then willbe discarded. To avoid this happening, you can instead configure a failure_transport:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        # after retrying, messages will be sent to the "failed" transport
        failure_transport: failed
 
        transports:
            # ... other transports
 
            failed: 'doctrine://default?queue_name=failed'

In this example, if handling a message fails 3 times (default maxretries),it will then be sent to the failed transport. While you _can usemessenger:consume failed to consume this like a normal transport, you'llusually want to manually view the messages in the failure transport and chooseto retry them:

# see all messages in the failure transport
$ php bin/console messenger:failed:show
 
# see details about a specific failure
$ php bin/console messenger:failed:show 20 -vv
 
# view and retry messages one-by-one
$ php bin/console messenger:failed:retry -vv
 
# retry specific messages
$ php bin/console messenger:failed:retry 20 30 --force
 
# remove a message without retrying it
$ php bin/console messenger:failed:remove 20

If the message fails again, it will be re-sent back to the failure transportdue to the normal retry rules. Once the maxretry has been hit, the message will be discarded permanently.

Transport Configuration

Messenger supports a number of different transport types, each with their ownoptions.

AMQP Transport

The amqp transport configuration looks like this:

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:[email protected]:5672/%2f/messages

To use Symfony's built-in AMQP transport, you need the AMQP PHP extension.

Note

By default, the transport will automatically create any exchanges, queues andbinding keys that are needed. That can be disabled, but some functionalitymay not work correctly (like delayed queues).

The transport has a number of other options, including ways to configurethe exchange, queues binding keys and more. See the documentation onConnection.

You can also configure AMQP-specific settings on your message by addingAmqpStamp toyour Envelope:

use Symfony\Component\Messenger\Transport\AmqpExt\AmqpStamp;
// ...
 
$attributes = [];
$bus->dispatch(new SmsNotification(), [
    new AmqpStamp('custom-routing-key', AMQP_NOPARAM, $attributes)
]);

Caution

The consumers do not show up in an admin panel as this transport does not rely on\AmqpQueue::consume() which is blocking. Having a blocking receiver makesthe —time-limit/—memory-limit options of the messenger:consume command as well asthe messenger:stop-workers command inefficient, as they all rely on the fact thatthe receiver returns immediately no matter if it finds a message or not. The consumeworker is responsible for iterating until it receives a message to handle and/or until oneof the stop conditions is reached. Thus, the worker's stop logic cannot be reached if itis stuck in a blocking call.

Doctrine Transport

New in version 4.3: The Doctrine transport was introduced in Symfony 4.3.

The Doctrine transport can be used to store messages in a database table.

# .env
MESSENGER_TRANSPORT_DSN=doctrine://default

The format is doctrine://<connection_name>, in case you have multiple connectionsand want to use one other than the "default". The transport will automatically createa table named messenger_messages (this is configurable) when the transport isfirst used. You can disable that with the auto_setup option and set the tableup manually by calling the messenger:setup-transports command.

Tip

To avoid tools like Doctrine Migrations from trying to remove this table becauseit's not part of your normal schema, you can set the schema_filter option:

# config/packages/doctrine.yaml
doctrine:
    dbal:
        schema_filter: '~^(?!messenger_messages)~'

The transport has a number of options:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_high: "%env(MESSENGER_TRANSPORT_DSN)%?queue_name=high_priority"
            async_normal:
                dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
                options:
                    queue_name: normal_priority

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
 
    <framework:config>
        <framework:messenger>
            <framework:transport name="async_priority_high" dsn="%env(MESSENGER_TRANSPORT_DSN)%?queue_name=high_priority"/>
            <framework:transport name="async_priority_low" dsn="%env(MESSENGER_TRANSPORT_DSN)%">
                <framework:option queue_name="normal_priority"/>
            </framework:transport>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'transports' => [
            'async_priority_high' => 'dsn' => '%env(MESSENGER_TRANSPORT_DSN)%?queue_name=high_priority',
            'async_priority_low' => [
                'dsn' => '%env(MESSENGER_TRANSPORT_DSN)%',
                'options' => [
                    'queue_name' => 'normal_priority'
                ]
            ],
        ],
    ],
]);

Options defined under options take precedence over ones defined in the DSN.

Option	Description	Default
table_name	Name of the table	messenger_messages
queue_name	Name of the queue (a column in thetable, to use one table formultiple transports)	default
redeliver_timeout	Timeout before retrying a messagethat's in the queue but in the"handling" state (if a worker diedfor some reason, this will occur,eventually you should retry themessage) - in seconds.	3600
auto_setup	Whether the table should be createdautomatically during send / get.	true

Redis Transport

New in version 4.3: The Redis transport was introduced in Symfony 4.3.

The Redis transport uses streams to queue messages.

# .env
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# Full DSN Example
MESSENGER_TRANSPORT_DSN=redis://[email protected]:6379/messages/symfony/consumer?auto_setup=true&serializer=1

To use the Redis transport, you will need the Redis PHP extension (^4.3) anda running Redis server (^5.0).

Caution

The Redis transport does not support "delayed" messages.

A number of options can be configured via the DSN or via the options keyunder the transport in messenger.yaml:

Option	Description	Default
stream	The Redis stream name	messages
group	The Redis consumer group name	symfony
consumer	Consumer name used in Redis	consumer
auto_setup	Create the Redis group automatically?	true
auth	The Redis password
serializer	How to serialize the final payloadin Redis (the`Redis::OPT_SERIALIZER` option)	`Redis::SERIALIZER_PHP`

In Memory Transport

New in version 4.3: The in-memory transport was introduced in Symfony 4.3.

The in-memory transport does not actually delivery messages. Instead, itholds them in memory during the request, which can be useful for testing.For example, if you have an async_priority_normal transport, you couldoverride it in the test environment to use this transport:

# config/packages/test/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_normal: 'in-memory:///'

Then, while testing, messages will not be delivered to the real transport.Even better, in a test, you can check that exactly one message was sentduring a request:

// tests/DefaultControllerTest.php
namespace App\Tests;
 
use Symfony\Bundle\FrameworkBundle\Test\WebTestCase;
use Symfony\Component\Messenger\Transport\InMemoryTransport;
 
class DefaultControllerTest extends WebTestCase
{
    public function testSomething()
    {
        $client = static::createClient();
        // ...
 
        $this->assertSame(200, $client->getResponse()->getStatusCode());
 
        /* @var InMemoryTransport $transport */
        $transport = self::$container->get('messenger.transport.async_priority_normal');
        $this->assertCount(1, $transport->get());
    }
}

Note

All in-memory transports will be reset automatically after each test intest classes extendingKernelTestCaseor WebTestCase.

Serializing Messages

New in version 4.3: The default serializer changed in 4.3 from the Symfony serializer to thenative PHP serializer. Existing applications should configure their transportsto use the Symfony serializer to avoid losing already-queued messages afterupgrading.

When messages are sent to (and received from) a transport, they're serializedusing PHP's native serialize() & unserialize() functions. You can changethis globally (or for each transport) to a service that implementsSerializerInterface:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        serializer:
            default_serializer: messenger.transport.symfony_serializer
            symfony_serializer:
                format: json
                context: { }
 
        transports:
            async_priority_normal:
                dsn: # ...
                serializer: messenger.transport.symfony_serializer

The messenger.transport.symfonyserializer is a built-in service that usesthe [_Serializer component]($e4d319ced595d34c.md) and can be configured in a few ways.If you do choose to use the Symfony serializer, you can control the contexton a case-by-case basis via the SerializerStamp(see Envelopes & Stamps).

Customizing Handlers

Manually Configuring Handlers

Symfony will normally find and register your handler automatically.But, you can also configure a handler manually - and pass it some extra config -by tagging the handler service with messenger.message_handler

YAML

# config/services.yaml
services:
    App\MessageHandler\SmsNotificationHandler:
        tags: [messenger.message_handler]
 
        # or configure with options
        tags:
            -
                name: messenger.message_handler
                # only needed if can't be guessed by type-hint
                handles: App\Message\SmsNotification
 
                # options returned by getHandledMessages() are supported here

XML

<!-- config/services.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd">
 
    <services>
        <service id="App\MessageHandler\SmsNotificationHandler">
           <tag name="messenger.message_handler"/>
        </service>
    </services>
</container>

PHP

// config/services.php
use App\MessageHandler\SmsNotificationHandler;

$container->register(SmsNotificationHandler::class)
    ->addTag('messenger.message_handler');

Handler Subscriber & Options

A handler class can handle multiple messages or configure itself by implementingMessageSubscriberInterface:

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

use App\Message\OtherSmsNotification;
use App\Message\SmsNotification;
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;

class SmsNotificationHandler implements MessageSubscriberInterface
{
    public function __invoke(SmsNotification $message)
    {
        // ...
    }

    public function handleOtherSmsNotification(OtherSmsNotification $message)
    {
        // ...
    }

    public static function getHandledMessages(): iterable
    {
        // handle this message on __invoke
        yield SmsNotification::class;

        // also handle this message on handleOtherSmsNotification
        yield OtherSmsNotification::class => [
            'method' => 'handleOtherSmsNotification',
            //'priority' => 0,
            //'bus' => 'messenger.bus.default',
        ];
    }
}

Binding Handlers to Different Transports

Each message can have multiple handlers, and when a message is consumedall of its handlers are called. But you can also configure a handler to onlybe called when it's received from a specific transport. This allows you tohave a single message where each handler is called by a different "worker"that's consuming a different transport.

Suppose you have an UploadedImage message with two handlers:

ThumbnailUploadedImageHandler: you want this to be handled bya transport called image_transport
NotifyAboutNewUploadedImageHandler: you want this to be handledby a transport called async_priority_normalTo do this, add the from_transport option to each handler. For example:

// src/MessageHandler/ThumbnailUploadedImageHandler.php
namespace App\MessageHandler;

use App\Message\UploadedImage;
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;

class ThumbnailUploadedImageHandler implements MessageSubscriberInterface
{
    public function __invoke(UploadedImage $uploadedImage)
    {
        // do some thumbnailing
    }

    public static function getHandledMessages(): iterable
    {
        yield UploadedImage::class => [
            'from_transport' => 'image_transport',
        ];
    }
}

And similarly:

// src/MessageHandler/NotifyAboutNewUploadedImageHandler.php
// ...

class NotifyAboutNewUploadedImageHandler implements MessageSubscriberInterface
{
    // ...

    public static function getHandledMessages(): iterable
    {
        yield UploadedImage::class => [
            'from_transport' => 'async_priority_normal',
        ];
    }
}

Then, make sure to "route" your message to both transports:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_normal: # ...
            image_transport: # ...

        routing:
            # ...
            'App\Message\UploadedImage': [image_transport, async_priority_normal]

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">

    <framework:config>
        <framework:messenger>
            <framework:transport name="async_priority_normal" dsn="..."/>
            <framework:transport name="image_transport" dsn="..."/>

            <framework:routing message-class="App\Message\UploadedImage">
                <framework:sender service="image_transport"/>
                <framework:sender service="async_priority_normal"/>
            </framework:routing>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'transports' => [
            'async_priority_normal' => '...',
            'image_transport' => '...',
        ],
        'routing' => [
            'App\Message\UploadedImage' => ['image_transport', 'async_priority_normal']
        ]
    ],
]);

That's it! You can now consume each transport:

# will only call ThumbnailUploadedImageHandler when handling the message
$ php bin/console messenger:consume image_transport -vv

$ php bin/console messenger:consume async_priority_normal -vv

Caution

If a handler does not have fromtransport config, it will be executedon _every transport that the message is received from.

Extending Messenger

Envelopes & Stamps

A message can be any PHP object. Sometimes, you may need to configure somethingextra about the message - like the way it should be handled inside Amqp or addinga delay before the message should be handled. You can do that by adding a "stamp"to your message:

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;

public function index(MessageBusInterface $bus)
{
    $bus->dispatch(new SmsNotification('...'), [
        // wait 5 seconds before processing
        new DelayStamp(5000)
    ]);

    // or explicitly create an Envelope
    $bus->dispatch(new Envelope(new SmsNotification('...'), [
        new DelayStamp(5000)
    ]));

    // ...
}

Internally, each message is wrapped in an Envelope, which holds the messageand stamps. You can create this manually or allow the message bus to do it. Thereare a variety of different stamps for different purposes and they're used internallyto track information about a message - like the message bus that's handling itor if it's being retried after failure.

Middleware

What happens when you dispatch a message to a message bus depends on itscollection of middleware (and their order). By default, the middleware configuredfor each bus looks like this:

add_bus_name_stamp_middleware - adds a stamp to record which bus thismessage was dispatched into;
dispatchafter_current_bus- see [_Transactional Messages: Handle New Messages After Handling is Done](https://symfony.com/doc/current/messenger/message-recorder.html);
failed_message_processing_middleware - processes messages that are beingretried via the failure transport to makethem properly function as if they were being received from their original transport;
Your own collection of middleware;
send_message - if routing is configured for the transport, this sendsmessages to that transport and stops the middleware chain;
handle_message - calls the message handler(s) for the given message.

Note

These middleware names are actually shortcuts names. The real service idsare prefixed with messenger.middleware..

You can add your own middleware to this list, or completely disable the defaultmiddleware and only include your own:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            messenger.bus.default:
                middleware:
                    # service ids that implement Symfony\Component\Messenger\Middleware
                    - 'App\Middleware\MyMiddleware'
                    - 'App\Middleware\AnotherMiddleware'

                #default_middleware: false

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">

    <framework:config>
        <framework:messenger>
            <framework:middleware id="App\Middleware\MyMiddleware"/>
            <framework:middleware id="App\Middleware\AnotherMiddleware"/>
            <framework:bus name="messenger.bus.default" default-middleware="false"/>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'buses' => [
            'messenger.bus.default' => [
                'middleware' => [
                    'App\Middleware\MyMiddleware',
                    'App\Middleware\AnotherMiddleware',
                ],
                'default_middleware' => false,
            ],
        ],
    ],
]);

Note

If a middleware service is abstract, a different instance of the service willbe created per bus.

Middleware for Doctrine

If you use Doctrine in your app, a number of optional middleware exist that youmay want to use:

YAML

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            command_bus:
                middleware:
                    # each time a message is handled, the Doctrine connection
                    # is "pinged" and reconnected if it's closed. Useful
                    # if your workers run for a long time and the database
                    # connection is sometimes lost
                    - doctrine_ping_connection

                    # After handling, the Doctrine connection is closed,
                    # which can free up database connections in a worker,
                    # instead of keeping them open forever
                    - doctrine_close_connection

                    # wraps all handlers in a single Doctrine transaction
                    # handlers do not need to call flush() and an error
                    # in any handler will cause a rollback
                    - doctrine_transaction

                    # or pass a different entity manager to any
                    #- doctrine_transaction: ['custom']

XML

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">

    <framework:config>
        <framework:messenger>
            <framework:bus name="command_bus">
                <framework:middleware id="doctrine_transaction"/>
                <framework:middleware id="doctrine_ping_connection"/>
                <framework:middleware id="doctrine_close_connection"/>

                <!-- or pass a different entity manager to any -->
                <!--
                <framework:middleware id="doctrine_transaction">
                    <framework:argument>custom</framework:argument>
                </framework:middleware>
                -->
            </framework:bus>
        </framework:messenger>
    </framework:config>
</container>

PHP

// config/packages/messenger.php
$container->loadFromExtension('framework', [
    'messenger' => [
        'buses' => [
            'command_bus' => [
                'middleware' => [
                    'doctrine_transaction',
                    'doctrine_ping_connection',
                    'doctrine_close_connection',
                    // Using another entity manager
                    ['id' => 'doctrine_transaction', 'arguments' => ['custom']],
                ],
            ],
        ],
    ],
]);

Messenger Events

In addition to middleware, Messenger also dispatches several events. You cancreate an event listener to hook into various partsof the process. For each, the event class is the event name:

Multiple Buses, Command & Event Buses

Messenger gives you a single message bus service by default. But, you can configureas many as you want, creating "command", "query" or "event" buses and controllingtheir middleware. See Multiple Buses.

Messenger

Messenger: Sync & Queued Message Handling

Installation

Creating a Message & Handler

Dispatching the Message

Transports: Async/Queued Messages

Routing Messages to a Transport

Doctrine Entities in Messages

Handling Messages Synchronously

Creating your Own Transport

Consuming Messages (Running the Worker)

Deploying to Production

Prioritized Transports

Supervisor Configuration

Retries & Failures

Avoiding Retrying

Saving & Retrying Failed Messages

Transport Configuration

AMQP Transport

Doctrine Transport

Redis Transport

In Memory Transport

Serializing Messages

Customizing Handlers

Manually Configuring Handlers

Handler Subscriber & Options

Binding Handlers to Different Transports

Extending Messenger

Envelopes & Stamps

Middleware

Middleware for Doctrine

Messenger Events

Multiple Buses, Command & Event Buses

Learn more