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:

1
2
3
4
5
6
7
8
9

# 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

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

<!-- 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>

1
2
3
4
5
6
7
8
9

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $framework->messenger()
        // async is whatever name you gave your transport above
        ->routing('App\Message\SmsNotification')->senders(['async'])
    ;
};

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, i.e. synchronously.

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

1
2
3
4
5
6
7
8
9

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

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

<!-- 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>

1
2
3
4
5
6
7
8
9
10

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();
    // route all messages that extend this example base class or interface
    $messenger->routing('App\Message\AbstractAsyncMessage')->senders(['async']);
    $messenger->routing('App\Message\AsyncMessageInterface')->senders(['async']);
    $messenger->routing('My\Message\ToBeSentToTwoSenders')->senders(['async', 'audit']);
};

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 (otherwise you might see errors related to the Entity Manager):

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

// src/Message/NewUserWelcomeEmail.php
namespace App\Message;

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
24

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

use App\Message\NewUserWelcomeEmail;
use App\Repository\UserRepository;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
class NewUserWelcomeEmailHandler
{
    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:

1
2
3
4
5
6
7
8
9
10

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            # ... other transports

            sync: 'sync://'

        routing:
            App\Message\SmsNotification: sync

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

<!-- 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>

1
2
3
4
5
6
7
8
9
10
11

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    // ... other transports

    $messenger->transport('sync')->dsn('sync://');
    $messenger->routing('App\Message\SmsNotification')->senders(['sync']);
};

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 a Process Manager like Supervisor or systemd 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 or systemd.

常に 1 つ以上の「ワーカー」を実行する必要があります。これを行うには、Supervisor または systemd などのプロセス制御システムを使用します。

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 the process manager will create a new process). There are also other options like --memory-limit=128M and --time-limit=3600.

一部のサービス (Doctrine の EntityManager など) は、時間の経過とともにより多くのメモリを消費します。したがって、ワーカーを永久に実行できるようにする代わりに、フラグのようなメッセンジャー:consume --limit=10 を使用して、終了する前に 10 個のメッセージのみを処理するようにワーカーに指示します (その後、プロセス マネージャーが新しいプロセスを作成します)。 --memory-limit=128M や --time-limit=3600 などの他のオプションもあります。

Stopping Workers That Encounter Errors

If a worker dependency like your database server is down, or timeout is reached, you can try to add reconnect logic, or just quit the worker if it receives too many errors with the --failure-limit option of the messenger:consume command.

データベース サーバーなどのワーカーの依存関係がダウンしている場合、またはタイムアウトに達した場合は、再接続ロジックを追加するか、messenger:consume コマンドの --failure-limit オプションでエラーが多すぎる場合はワーカーを終了することができます。

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 deployment. This will signal to each worker that it should finish the message it's currently handling and should shut down gracefully. Then, the process manager will create new worker processes. The command uses the app cache internally - so make sure this is configured to use an adapter you like.

デプロイするたびに、新しくデプロイされたコードが表示されるように、すべてのワーカー プロセスを再起動する必要があります。これを行うには、messenger:stop-workerson デプロイメントを実行します。これは、現在処理しているメッセージを終了し、正常にシャットダウンする必要があることを各ワーカーに通知します。次に、プロセス マネージャが新しいワーカー プロセスを作成します。このコマンドは appcache を内部的に使用するため、好きなアダプターを使用するように構成されていることを確認してください。

Use the Same Cache Between Deploys

If your deploy strategy involves the creation of new target directories, you should set a value for the cache.prefix.seed configuration option in order to use the same cache namespace between deployments. Otherwise, the cache.app pool will use the value of the kernel.project_dir parameter as base for the namespace, which will lead to different namespaces each time a new deployment is made.

展開戦略に新しいターゲット ディレクトリの作成が含まれる場合は、展開間で同じキャッシュ名前空間を使用するために、cache.prefix.seed 構成オプションの値を設定する必要があります。それ以外の場合、cache.app プールはカーネルの値を使用します。 project_dir パラメーターを名前空間のベースとして使用します。これにより、新しい展開が行われるたびに異なる名前空間が作成されます。

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:

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

# 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

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

<!-- 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)%">
                <framework:options>
                    <framework:queue>
                        <framework:name>Queue</framework:name>
                    </framework:queue>
                </framework:options>
            </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>

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

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $messenger->transport('async_priority_high')
        ->dsn(env('MESSENGER_TRANSPORT_DSN'))
        ->options(['queue_name' => 'high']);

    $messenger->transport('async_priority_low')
        ->dsn(env('MESSENGER_TRANSPORT_DSN'))
        ->options(['queue_name' => 'low']);

    $messenger->routing('App\Message\SmsNotification')->senders(['async_priority_low']);
    $messenger->routing('App\Message\NewUserWelcomeEmail')->senders(['async_priority_high']);
};

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.

Limit Consuming to Specific Queues ¶

Some transports (notably AMQP) have the concept of exchanges and queues. A Symfony transport is always bound to an exchange. By default, the worker consumes from all queues attached to the exchange of the specified transport. However, there are use cases to want a worker to only consume from specific queues.

You can limit the worker to only process messages from specific queue(s):

1
2
3
4

$ php bin/console messenger:consume my_transport --queues=fasttrack

# you can pass the --queues option more than once to process multiple queues
$ php bin/console messenger:consume my_transport --queues=fasttrack1 --queues=fasttrack2

Checking the Number of Queued Messages Per Transport ¶

Run the messenger:stats command to know how many messages are in the "queues" of some or all transports:

1
2
3
4
5

# displays the number of queued messages in all transports
$ php bin/console messenger:stats

# shows stats only for some transports
$ php bin/console messenger:stats my_transport_name other_transport_name

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
9
10

;/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
startsecs=0
autostart=true
autorestart=true
startretries=10
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.

If you use the Redis Transport, note that each worker needs a unique consumer name to avoid the same message being handled by multiple workers. One way to achieve this is to set an environment variable in the Supervisor configuration file, which you can then refer to in messenger.yaml (see Redis section above):

1

environment=MESSENGER_CONSUMER_NAME=%(program_name)s_%(process_num)02d

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.

1
2

[program:x]
stopwaitsecs=20

Systemd Configuration ¶

While Supervisor is a great tool, it has the disadvantage that you need system access to run it. Systemd has become the standard on most Linux distributions, and has a good alternative called user services.

Systemd user service configuration files typically live in a ~/.config/systemd/user directory. For example, you can create a new messenger-worker.service file. Or a messenger-worker@.service file if you want more instances running at the same time:

1
2
3
4
5
6
7
8
9
10

[Unit]
Description=Symfony messenger-consume %i

[Service]
ExecStart=php /path/to/your/app/bin/console messenger:consume async --time-limit=3600
Restart=always
RestartSec=30

[Install]
WantedBy=default.target

Now, tell systemd to enable and start one worker:

1
2
3
4
5
6

$ systemctl --user enable messenger-worker@1.service
$ systemctl --user start messenger-worker@1.service

# to enable and start 20 workers
$ systemctl --user enable messenger-worker@{1..20}.service
$ systemctl --user start messenger-worker@{1..20}.service

If you change your service config file, you need to reload the daemon:

1

$ systemctl --user daemon-reload

To restart all your consumers:

1

$ systemctl --user restart messenger-consume@*.service

The systemd user instance is only started after the first login of the particular user. Consumer often need to start on system boot instead. Enable lingering on the user to activate that behavior:

1

$ loginctl enable-linger <your-username>

Logs are managed by journald and can be worked with using the journalctl command:

1
2
3
4
5
6
7
8

# follow logs of consumer nr 11
$ journalctl -f --user-unit messenger-consume@11.service

# follow logs of all consumers
$ journalctl -f --user-unit messenger-consume@*

# follow all logs from your user services
$ journalctl -f _UID=$UID

See the systemd docs for more details.

1

$ sudo usermod -a -G systemd-journal <your-username>

Stateless Worker ¶

PHP is designed to be stateless, there are no shared resources across different requests. In HTTP context PHP cleans everything after sending the response, so you can decide to not take care of services that may leak memory.

On the other hand, it's common for workers to process messages sequentially in long-running CLI processes which don't finish after processing a single message. Beware about service states to prevent information and/or memory leakage as Symfony will inject the same instance of a service in all messages, preserving the internal state of the services.

However, certain Symfony services, such as the Monolog fingers crossed handler, leak by design. Symfony provides a service reset feature to solve this problem. When resetting the container automatically between two messages, Symfony looks for any services implementing ResetInterface (including your own services) and calls their reset() method so they can clean their internal state.

If a service is not stateless and you want to reset its properties after each message, then the service must implement ResetInterface where you can reset the properties in the reset() method.

If you don't want to reset the container, add the --no-reset option when running the messenger:consume command.

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.

Forcing Retrying ¶

Sometimes handling a message must fail in a way that you know is temporary and must be retried. If you throw RecoverableMessageHandlingException, the message will always be retried infinitely and max_retries setting will be ignored.

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:

1
2
3
4
5
6
7
8
9
10

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

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

<!-- 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>
        <!-- after retrying, messages will be sent to the "failed" transport -->
        <framework:messenger failure-transport="failed">
            <!-- ... other transports -->

            <framework:transport name="failed" dsn="doctrine://default?queue_name=failed"/>
        </framework:messenger>
    </framework:config>
</container>

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

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    // after retrying, messages will be sent to the "failed" transport
    $messenger->failureTransport('failed');

    // ... other transports

    $messenger->transport('failed')
        ->dsn('doctrine://default?queue_name=failed');
};

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
15
16
17
18
19
20
21
22
23
24
25
26

# see all messages in the failure transport with a default limit of 50
$ php bin/console messenger:failed:show

# see the 10 first messages
$ php bin/console messenger:failed:show --max=10

# see only MyClass messages
$ php bin/console messenger:failed:show --class-filter='MyClass'

# see the number of messages by message class
$ php bin/console messenger:failed:show --stats

# 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

# remove messages without retrying them and show each message before removing it
$ php bin/console messenger:failed:remove 20 30 --show-messages

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.

Multiple Failed Transports ¶

Sometimes it is not enough to have a single, global failed transport configured because some messages are more important than others. In those cases, you can override the failure transport for only specific transports:

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

# config/packages/messenger.yaml
framework:
    messenger:
        # after retrying, messages will be sent to the "failed" transport
        # by default if no "failed_transport" is configured inside a transport
        failure_transport: failed_default

        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                failure_transport: failed_high_priority

            # since no failed transport is configured, the one used will be
            # the global "failure_transport" set
            async_priority_low:
                dsn: 'doctrine://default?queue_name=async_priority_low'

            failed_default: 'doctrine://default?queue_name=failed_default'
            failed_high_priority: 'doctrine://default?queue_name=failed_high_priority'

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

<!-- 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>
        <!-- after retrying, messages will be sent to the "failed" transport
        by default if no "failed-transport" is configured inside a transport -->
        <framework:messenger failure-transport="failed_default">
            <framework:transport name="async_priority_high" dsn="%env(MESSENGER_TRANSPORT_DSN)%" failure-transport="failed_high_priority"/>
            <!-- since no "failed_transport" is configured, the one used will be
            the global "failed_transport" set -->
            <framework:transport name="async_priority_low" dsn="doctrine://default?queue_name=async_priority_low"/>

            <framework:transport name="failed_default" dsn="doctrine://default?queue_name=failed_default"/>
            <framework:transport name="failed_high_priority" dsn="doctrine://default?queue_name=failed_high_priority"/>
        </framework:messenger>
    </framework:config>
</container>

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

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    // after retrying, messages will be sent to the "failed" transport
    // by default if no "failure_transport" is configured inside a transport
    $messenger->failureTransport('failed_default');

    $messenger->transport('async_priority_high')
        ->dsn(env('MESSENGER_TRANSPORT_DSN'))
        ->failureTransport('failed_high_priority');

    // since no failed transport is configured, the one used will be
    // the global failure_transport set
   $messenger->transport('async_priority_low')
        ->dsn('doctrine://default?queue_name=async_priority_low');

   $messenger->transport('failed_default')
        ->dsn('doctrine://default?queue_name=failed_default');

   $messenger->transport('failed_high_priority')
        ->dsn('doctrine://default?queue_name=failed_high_priority');
};

If there is no failure_transport defined globally or on the transport level, the messages will be discarded after the number of retries.

The failed commands have an optional option --transport to specify the failure_transport configured at the transport level.

1
2
3
4
5
6
7
8

# see all messages in "failure_transport" transport
$ php bin/console messenger:failed:show --transport=failure_transport

# retry specific messages from "failure_transport"
$ php bin/console messenger:failed:retry 20 30 --transport=failure_transport --force

# remove a message without retrying it from "failure_transport"
$ php bin/console messenger:failed:remove 20 --transport=failure_transport

AMQP Transport ¶

The AMQP transport uses the AMQP PHP extension to send messages to queues like RabbitMQ. Install it by running:

1

$ composer require symfony/amqp-messenger

The AMQP transport DSN may looks like this:

1
2
3
4
5

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages

# or use the AMQPS protocol
MESSENGER_TRANSPORT_DSN=amqps://guest:guest@localhost/%2f/messages

If you want to use TLS/SSL encrypted AMQP, you must also provide a CA certificate. Define the certificate path in the amqp.cacert PHP.ini setting (e.g. amqp.cacert = /etc/ssl/certs) or in the cacert parameter of the DSN (e.g amqps://localhost?cacert=/etc/ssl/certs/).

The default port used by TLS/SSL encrypted AMQP is 5671, but you can overwrite it in the port parameter of the DSN (e.g. amqps://localhost?cacert=/etc/ssl/certs/&port=12345).

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

The transport has a number of options:

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 ¶

The Doctrine transport can be used to store messages in a database table. Install it by running:

1

$ composer require symfony/doctrine-messenger

The Doctrine transport DSN may looks like this:

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.

Or, to create the table yourself, set the auto_setup option to false and generate a migration.

1
2
3
4

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

1
2

<!-- config/packages/doctrine.xml -->
<doctrine:dbal schema-filter="~^(?!messenger_messages)~"/>

1
2
3
4
5
6
7
8

# config/packages/doctrine.php
$container->loadFromExtension('doctrine', [
    'dbal' => [
        'schema_filter'  => '~^(?!messenger_messages)~',
        // ...
    ],
    // ...
]);

The transport has a number of options:

When using PostgreSQL, you have access to the following options to leverage the LISTEN/NOTIFY feature. This allow for a more performant approach than the default polling behavior of the Doctrine transport because PostgreSQL will directly notify the workers when a new message is inserted in the table.

Beanstalkd Transport ¶

The Beanstalkd transport sends messages directly to a Beanstalkd work queue. Install it by running:

1

$ composer require symfony/beanstalkd-messenger

The Beanstalkd transport DSN may looks like this:

1
2
3
4
5

# .env
MESSENGER_TRANSPORT_DSN=beanstalkd://localhost:11300?tube_name=foo&timeout=4&ttr=120

# If no port, it will default to 11300
MESSENGER_TRANSPORT_DSN=beanstalkd://localhost

The transport has a number of options:

Redis Transport ¶

The Redis transport uses streams to queue messages. This transport requires the Redis PHP extension (>=4.3) and a running Redis server (^5.0). Install it by running:

1

$ composer require symfony/redis-messenger

The Redis transport DSN may looks like this:

1
2
3
4
5
6
7
8

# .env
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# Full DSN Example
MESSENGER_TRANSPORT_DSN=redis://password@localhost:6379/messages/symfony/consumer?auto_setup=true&serializer=1&stream_max_entries=0&dbindex=0
# Redis Cluster Example
MESSENGER_TRANSPORT_DSN=redis://host-01:6379,redis://host-02:6379,redis://host-03:6379,redis://host-04:6379
# Unix Socket Example
MESSENGER_TRANSPORT_DSN=redis:///var/run/redis.sock

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

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

<!-- config/packages/test/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="in-memory://"/>
        </framework:messenger>
    </framework:config>
</container>

1
2
3
4
5
6
7
8
9

// config/packages/test/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $messenger->transport('async_priority_normal')
        ->dsn('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/Controller/DefaultControllerTest.php
namespace App\Tests\Controller;

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 = $this->getContainer()->get('messenger.transport.async_priority_normal');
        $this->assertCount(1, $transport->getSent());
    }
}

The transport has a number of options:

serialize (boolean, default: false)

Whether to serialize messages or not. This is useful to test an additional layer, especially when you use your own message serializer.

メッセージをシリアライズするかどうか。これは、特に独自のメッセージシリアライザーを使用する場合に、追加レイヤーをテストするのに役立ちます。

Amazon SQS ¶

The Amazon SQS transport is perfect for applications hosted on AWS. Install it by running:

1

$ composer require symfony/amazon-sqs-messenger

The SQS transport DSN may looks like this:

1
2
3

# .env
MESSENGER_TRANSPORT_DSN=https://sqs.eu-west-3.amazonaws.com/123456789012/messages?access_key=AKIAIOSFODNN7EXAMPLE&secret_key=j17M97ffSVoKI0briFoo9a
MESSENGER_TRANSPORT_DSN=sqs://localhost:9494/messages?sslmode=disable

The transport has a number of options:

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:

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

# 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

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

<!-- 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:serializer default-serializer="messenger.transport.symfony_serializer">
                <framework:symfony-serializer format="json">
                    <framework:context/>
                </framework:symfony-serializer>
            </framework:serializer>

            <framework:transport name="async_priority_normal" dsn="..." serializer="messenger.transport.symfony_serializer"/>
        </framework:messenger>
    </framework:config>
</container>

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

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $messenger->serializer()
        ->defaultSerializer('messenger.transport.symfony_serializer')
        ->symfonySerializer()
            ->format('json')
            ->context('foo', 'bar');

    $messenger->transport('async_priority_normal')
        ->dsn('...')
        ->serializer('messenger.transport.symfony_serializer');
};

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

Configuring Handlers Using Attributes ¶

You can configure your handler by passing options to the attribute:

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

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

use App\Message\OtherSmsNotification;
use App\Message\SmsNotification;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler(fromTransport: 'async', priority: 10)]
class SmsNotificationHandler
{
    public function __invoke(SmsNotification $message)
    {
        // ...
    }
}

Possible options to configure with the attribute are:

• bus
  バス
• fromTransport
  fromTransport
• handles
  ハンドル
• method
  方法
• priority
  優先順位

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

1
2
3
4
5
6
7
8
9
10
11

# 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

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

<!-- 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">
             <!-- handles is only needed if it can't be guessed by type-hint -->
             <tag name="messenger.message_handler"
                  handles="App\Message\SmsNotification"/>
        </service>
    </services>
</container>

1
2
3
4
5
6
7
8
9

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

$container->register(SmsNotificationHandler::class)
    ->addTag('messenger.message_handler', [
        // only needed if can't be guessed by type-hint
        'handles' => SmsNotification::class,
    ]);

Possible options to configure with tags are:

• bus
  バス
• from_transport
  from_transport
• handles
  ハンドル
• method
  方法
• priority
  優先順位

Handling Multiple Messages ¶

A single handler class can handle multiple messages. For that add the #AsMessageHandler attribute to all the handling methods:

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

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

use App\Message\OtherSmsNotification;
use App\Message\SmsNotification;

class SmsNotificationHandler
{
    #[AsMessageHandler]
    public function handleSmsNotification(SmsNotification $message)
    {
        // ...
    }

    #[AsMessageHandler]
    public function handleOtherSmsNotification(OtherSmsNotification $message)
    {
        // ...
    }
}

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
  ThumbnailUploadedImageHandler: これを image_transport というトランスポートで処理したい
• NotifyAboutNewUploadedImageHandler: you want this to be handled by a transport called async_priority_normal
  NotifyAboutNewUploadedImageHandler: これを 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

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

use App\Message\UploadedImage;

#[AsMessageHandler(from_transport: 'image_transport')]
class ThumbnailUploadedImageHandler
{
    public function __invoke(UploadedImage $uploadedImage)
    {
        // do some thumbnailing
    }
}

And similarly:

1
2
3
4
5
6
7
8

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

#[AsMessageHandler(from_transport: 'async_priority_normal')]
class NotifyAboutNewUploadedImageHandler
{
    // ...
}

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

1
2
3
4
5
6
7
8
9
10

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

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

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

<!-- 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>

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

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $messenger->transport('async_priority_normal')->dsn('...');
    $messenger->transport('image_transport')->dsn('...');

    $messenger->routing('App\Message\UploadedImage')
        ->senders(['image_transport', 'async_priority_normal']);
};

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;
   add_bus_name_stamp_middleware - このメッセージがディスパッチされたバスを記録するスタンプを追加します。
2. dispatch_after_current_bus- see Transactional Messages: Handle New Messages After Handling is Done;
   dispatch_after_current_bus-「トランザクション メッセージ: 処理が完了した後に新しいメッセージを処理する」を参照してください。
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;
   failed_message_processing_middleware - 障害トランスポートを介して再試行されているメッセージを処理して、元のトランスポートから受信されたかのように適切に機能させます。
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;
   send_message - トランスポート用にルーティングが構成されている場合、これはメッセージをそのトランスポートに送信し、ミドルウェア チェーンを停止します。
6. handle_message - calls the message handler(s) for the given message.
   handle_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:

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

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            messenger.bus.default:
                # disable the default middleware
                default_middleware: false

                # and/or add your own
                middleware:
                    # service ids that implement Symfony\Component\Messenger\Middleware\MiddlewareInterface
                    - 'App\Middleware\MyMiddleware'
                    - 'App\Middleware\AnotherMiddleware'

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

<!-- 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>
            <!-- default-middleware: disable the default middleware -->
            <framework:bus name="messenger.bus.default" default-middleware="false"/>

            <!-- and/or add your own -->
            <framework:middleware id="App\Middleware\MyMiddleware"/>
            <framework:middleware id="App\Middleware\AnotherMiddleware"/>
        </framework:messenger>
    </framework:config>
</container>

1
2
3
4
5
6
7
8
9
10
11

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $bus = $messenger->bus('messenger.bus.default')
        ->defaultMiddleware(false);
    $bus->middleware()->id('App\Middleware\MyMiddleware');
    $bus->middleware()->id('App\Middleware\AnotherMiddleware');
};

Middleware for Doctrine ¶

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

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

# 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']

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

<!-- 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>

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

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $bus = $messenger->bus('command_bus');
    $bus->middleware()->id('doctrine_transaction');
    $bus->middleware()->id('doctrine_ping_connection');
    $bus->middleware()->id('doctrine_close_connection');
    // Using another entity manager
    $bus->middleware()->id('doctrine_transaction')
        ->arguments(['custom']);
};

Other Middlewares ¶

Add the router_context middleware if you need to generate absolute URLs in the consumer (e.g. render a template with links). This middleware stores the original request context (i.e. the host, the HTTP port, etc.) which is needed when building absolute URLs.

Add the validation middleware if you need to validate the message object using the Validator component before handling it. If validation fails, a ValidationFailedException will be thrown. The ValidationStamp can be used to configure the validation groups.

1
2
3
4
5
6
7
8

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            command_bus:
                middleware:
                    - router_context
                    - validation

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

<!-- 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="router_context"/>
                <framework:middleware id="validation"/>
            </framework:bus>
        </framework:messenger>
    </framework:config>
</container>

1
2
3
4
5
6
7
8
9
10

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework) {
    $messenger = $framework->messenger();

    $bus = $messenger->bus('command_bus');
    $bus->middleware()->id('router_context');
    $bus->middleware()->id('validation');
};

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
  WorkerStartedEvent
• WorkerMessageReceivedEvent
  WorkerMessageReceivedEvent
• SendMessageToTransportsEvent
  SendMessageToTransportsEvent
• WorkerMessageFailedEvent
  WorkerMessageFailedEvent
• WorkerMessageHandledEvent
  WorkerMessageHandledEvent
• WorkerRunningEvent
  WorkerRunningEvent
• WorkerStoppedEvent
  WorkerStoppedEvent