Messenger: Sync & Queued Message Handling¶

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:

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

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

Doctrine Entities in Messages¶

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// 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't be sent to any transport and will be handled immediately. In some cases (like when binding handlers to different transports), it's easier or more flexible to handle this explicitly: by creating a sync transport and "sending" messages there to be handled immediately:

Creating your Own Transport¶

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

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 a process control system like Supervisor.

Don't Let Workers Run Forever

Some services (like Doctrine's EntityManager) will consume more memory over time. So, instead of allowing your worker to run forever, use a flag like messenger:consume --limit=10 to tell your worker to only handle 10 messages before exiting (then Supervisor will create a new process). There are 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 so that they see the newly deployed code. To do this, run messenger:stop-workers on deploy. This will signal to each worker that it should finish the message it's currently handling and shut down gracefully. Then, Supervisor will create new worker processes. The command uses the app cache 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 handled before others. To make this possible, you can create multiple transports and route different messages to them. For example:

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

1

$ php bin/console messenger:consume async_priority_high async_priority_low

The worker will always first look for messages waiting on async_priority_high. If there 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) is always running (even if it closes due to failure, hitting a message limit or thanks to messenger:stop-workers). You can install it on Ubuntu, for example, via:

1

$ sudo apt-get install supervisor

Supervisor configuration files typically live in a /etc/supervisor/conf.d directory. For example, you can create a new messenger-worker.conf file there to make sure that 2 instances of messenger:consume are running at all times:

1
2
3
4
5
6
7
8

;/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 your config and start your workers:

1
2
3
4
5

$ sudo supervisorctl reread

$ sudo supervisorctl update

$ sudo supervisorctl start messenger-consume:*

See the Supervisor docs for more details.

Avoiding Retrying¶

Sometimes handling a message might fail in a way that you know is permanent and should not be retried. If you throw UnrecoverableMessageHandlingException, the message will not be retried.

Saving & Retrying Failed Messages¶

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

In this example, if handling a message fails 3 times (default max_retries), it will then be sent to the failed transport. While you can use messenger:consume failed to consume this like a normal transport, you'll usually want to manually view the messages in the failure transport and choose to retry them:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 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 transport due to the normal retry rules. Once the max retry has been hit, the message will be discarded permanently.

AMQP Transport¶

1

$ composer require symfony/amqp-messenger

The amqp transport configuration looks like this:

1
2

# .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.

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

You can also configure AMQP-specific settings on your message by adding AmqpStamp to your Envelope:

1
2
3
4
5
6
7

use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpStamp;
// ...

$attributes = [];
$bus->dispatch(new SmsNotification(), [
    new AmqpStamp('custom-routing-key', AMQP_NOPARAM, $attributes)
]);

Doctrine Transport¶

1

$ composer require symfony/doctrine-messenger

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

1
2

# .env
MESSENGER_TRANSPORT_DSN=doctrine://default

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

1
2
3
4

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

The transport has a number of options:

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

Redis Transport¶

1

$ composer require symfony/redis-messenger

The Redis transport uses streams to queue messages.

1
2
3
4
5
6

# .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&stream_max_entries=0&dbindex=0
# Unix Socket Example
MESSENGER_TRANSPORT_DSN=redis:///var/run/redis.sock

To use the Redis transport, you will need the Redis PHP extension (>=4.3) and a running Redis server (^5.0).

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

In Memory Transport¶

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

1
2
3
4
5

# 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 sent during a request:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// 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->getSent());
    }
}

Serializing Messages¶

When messages are sent to (and received from) a transport, they're serialized using PHP's native serialize() & unserialize() functions. You can change this globally (or for each transport) to a service that implements SerializerInterface:

The messenger.transport.symfony_serializer is a built-in service that uses the Serializer component and can be configured in a few ways. If you do choose to use the Symfony serializer, you can control the context on a case-by-case basis via the SerializerStamp (see Envelopes & Stamps).

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

Possible options to configure with tags are:

• bus
• from_transport
• handles
• method
• priority

Handler Subscriber & Options¶

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// 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 consumed all of its handlers are called. But you can also configure a handler to only be called when it's received from a specific transport. This allows you to have 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 by a transport called image_transport
• NotifyAboutNewUploadedImageHandler: you want this to be handled by a transport called async_priority_normal

To do this, add the from_transport option to each handler. For example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// 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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// 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:

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

1
2
3
4

# 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

Envelopes & Stamps¶

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

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 message and stamps. You can create this manually or allow the message bus to do it. There are a variety of different stamps for different purposes and they're used internally to track information about a message - like the message bus that's handling it or if it's being retried after failure.

Middleware¶

What happens when you dispatch a message to a message bus depends on its collection of middleware and their order. By default, the middleware configured for each bus looks like this:

1. add_bus_name_stamp_middleware - adds a stamp to record which bus this message was dispatched into;
2. dispatch_after_current_bus- see Transactional Messages: Handle New Messages After Handling is Done;
3. failed_message_processing_middleware - processes messages that are being retried via the failure transport to make them properly function as if they were being received from their original transport;
4. Your own collection of middleware;
5. send_message - if routing is configured for the transport, this sends messages to that transport and stops the middleware chain;
6. handle_message - calls the message handler(s) for the given message.

The middleware are executed when the message is dispatched but also again when a message is received via the worker (for messages that were sent to a transport to be handled asynchronously). Keep this in mind if you create your own middleware.

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

Middleware for Doctrine¶

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

Messenger Events¶

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

• WorkerStartedEvent
• WorkerMessageReceivedEvent
• SendMessageToTransportsEvent
• WorkerMessageFailedEvent
• WorkerMessageHandledEvent
• WorkerRunningEvent
• WorkerStoppedEvent