New in Symfony 5.4: Messenger Improvements

Configurable handlers with PHP attributes

Contributed by Alireza Mirsepassi in #43588

PHP attributes are a great way of adding metadata to PHP code. In Symfony we're adding the option of using PHP attributes to configure most things. That's why in Symfony 5.4 we're allowing to configure message handlers with attributes.

Instead of having to implement the MessageHandlerInterface, you can now add the AsMessageHandler attribute to any PHP class and use it as a message handler:

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)
    {
        // ...
    }
}

Worker metadata

Contributed by Oleg Krasavin in #42335

Currently there's no simple way to get worker metadata such as its transport name. In Symfony 5.4 we're improving this with the introduction of a new WorkerMetadata class which is accessible via $worker->getWorkerMetadata(). For example, inside a method of some listener/subscriber that handles Symfony Messenger events you could use something like this:

1
2
3
4
5
6
7
8
9

public function resetServices(WorkerRunningEvent $event): void
{
    $actualTransportName = $event->getWorker()->getWorkerMetadata()->getTransportName();
    if (!$event->isWorkerIdle() || !in_array($actualTransportName, $this->receiversName, true)) {
        return;
    }

    $this->servicesResetter->reset();
}

Reset container services between messages

Contributed by Grégoire Pineau in #41163 and #43322

Container services are not reset automatically when handling messages. This can be a problem for example with the Monolog fingers crossed handler. Since services are not reset, if the first message triggers an error, the next messages will log and ultimately overflow the buffer.

In Symfony 5.4 we're improving this situation with the option to automatically reset services after handling a message. To use this feature, set the new reset_on_message option to true in your messenger configuration:

1
2
3
4
5
6
7
8
9

# config/packages/messenger.yaml
framework:
    messenger:
        reset_on_message: true
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
            failed: 'doctrine://default?queue_name=failed'
            sync: 'sync://'

Handle messages in batches

Contributed by Nicolas Grekas in #43354

Sometimes, when using the Messenger component, you could handle multiple messages at once instead of processing them one by one. In Symfony 5.4 we've introduced a new BatchHandlerInterface that allows your handlers to process messages in batches.

Handlers implementing this interface should expect a new $ack optional argument to be provided when __invoke() is called. If you don't provide the $ack argument, the message is handled synchronously as usual. If you provide $ack, __invoke() is expected to buffer the message and its $ack function, and to return the number of pending messages in the batch. Here is what a batch handler might look like:

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

class MyBatchHandler implements BatchHandlerInterface
{
    use BatchHandlerTrait;

    public function __invoke(MyMessage $message, Acknowledger $ack = null)
    {
        return $this->handle($message, $ack);
    }

    private function process(array $jobs): void
    {
        foreach ($jobs as [$message, $ack]) {
            try {
                // [...] compute $result from $message
                $ack->ack($result);
            } catch (\Throwable $e) {
                $ack->nack($e);
            }
        }
    }
}

The size of the batch is controlled by BatchHandlerTrait::shouldFlush() (defaults to 10).