Architecture Message Bus

Messenger est un bus de messages — il sépare l'émetteur d'un message de son handler, avec la possibilité de traiter les messages de manière asynchrone.

Messenger repose sur un pattern simple : l'émetteur d'un message ne connaît pas ses destinataires. Il dépose le message dans le bus — le bus décide si le message est traité immédiatement par un handler ou envoyé dans une file pour traitement différé.

L'Envelope — le conteneur central

Tout message dispatché est immédiatement encapsulé dans une Envelope. C'est elle qui traverse les middlewares, pas le message brut. Elle accumule des stamps — des métadonnées ajoutées par chaque middleware.

Flux complet d'un dispatch

Quand tu appelles $bus->dispatch($message), voici ce qui se passe dans l'ordre :

1. Le message est wrappé dans une Envelope
2. L'Envelope traverse chaque middleware dans l'ordre de la chaîne
3. Si un routing est configuré → SendMessageMiddleware envoie au transport et s'arrête
4. Sinon → HandleMessageMiddleware appelle le ou les handlers correspondants
5. Les handlers retournent leurs résultats via des HandledStamp ajoutés à l'Envelope
6. Le bus retourne l'Envelope finale avec tous les stamps accumulés

Lance la simulation ci-dessous pour voir le flux en mode sync et async :

Envelope et Stamps

Tout message Messenger est enveloppé dans une Envelope — un conteneur immuable qui accumule des métadonnées sous forme de stamps.

L'Envelope est le conteneur immuable qui porte le message et ses métadonnées. Chaque middleware peut l'enrichir avec de nouveaux stamps — des objets légers qui documentent ce qui s'est passé, les décisions prises, les résultats obtenus.

API de l'Envelope

use Symfony\Component\Messenger\Envelope;

// Créer une Envelope
$envelope = Envelope::wrap($message);
$envelope = Envelope::wrap($message, [new DelayStamp(5000)]);

// Ajouter des stamps — retourne un CLONE (immuable)
$enriched = $envelope->with(new BusNameStamp('command.bus'));

// Supprimer tous les stamps d'un type — retourne un CLONE
$clean = $envelope->withoutAll(ValidationStamp::class);

// Récupérer le dernier stamp d'un type
$stamp = $envelope->last(HandledStamp::class); // ?HandledStamp

// Récupérer tous les stamps d'un type
$stamps = $envelope->all(HandledStamp::class); // array

// Récupérer le message sous-jacent
$message = $envelope->getMessage();

Les stamps principaux

Récupérer le résultat d'un handler

$envelope = $bus->dispatch(new FindUserQuery($id));

// Résultat du premier handler
$user = $envelope->last(HandledStamp::class)?->getResult();

// Résultats de tous les handlers (si plusieurs)
$results = array_map(
    fn(HandledStamp $s) => $s->getResult(),
    $envelope->all(HandledStamp::class)
);

Handlers — #[AsMessageHandler]

Un handler est une classe invokable (ou une méthode) décorée avec #[AsMessageHandler] — elle reçoit un message et agit dessus.

Depuis Symfony Messenger 6.2, MessageHandlerInterface n'existe plus. La seule façon de déclarer un handler est d'utiliser l'attribut #[AsMessageHandler]. Symfony détecte automatiquement le type du message depuis le paramètre de __invoke().

#[AsMessageHandler] — la forme canonique

use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class CreateUserHandler
{
    public function __construct(private readonly UserRepository $users) {}

    public function __invoke(CreateUserCommand $command): User
    {
        $user = User::create($command->email);
        $this->users->save($user);
        return $user;
    }
}

Paramètres de l'attribut

Handler avec méthode nommée

#[AsMessageHandler(handles: SendEmailCommand::class, method: 'onSendEmail')]
final class MailHandler
{
    public function onSendEmail(SendEmailCommand $cmd): void
    {
        // envoyer l'email
    }
}

Multiple handlers pour un même message

Plusieurs handlers peuvent gérer le même type de message. Ils sont tous appelés, dans l'ordre de priorité décroissant. Chacun ajoute un HandledStamp à l'Envelope.

#[AsMessageHandler(priority: 10)] // appelé en premier
final class AuditHandler
{
    public function __invoke(UserCreatedEvent $event): void { /* log */ }
}

#[AsMessageHandler(priority: 5)] // appelé ensuite
final class WelcomeEmailHandler
{
    public function __invoke(UserCreatedEvent $event): void { /* email */ }
}

HandlerFailedException

Si un handler lève une exception, Messenger la wrappe dans une HandlerFailedException. Elle implémente WrappedExceptionsInterface — tu peux accéder aux exceptions originales :

use Symfony\Component\Messenger\Exception\HandlerFailedException;

try {
    $bus->dispatch(new CreateUserCommand($email));
} catch (HandlerFailedException $e) {
    foreach ($e->getWrappedExceptions() as $inner) {
        // traiter chaque exception de handler
    }
}

Transports et routing

Un transport est une infrastructure de file de messages — AMQP, Doctrine, Redis, In-Memory. Le routing détermine quel message va sur quel transport.

Un transport est une infrastructure persistante pour les messages asynchrones — une file AMQP, une table Doctrine, un stream Redis. Quand un message est routé vers un transport, il n'est pas traité immédiatement : il attend qu'un Worker le consomme.

TransportInterface

interface TransportInterface extends ReceiverInterface, SenderInterface
{
    // SenderInterface
    public function send(Envelope $envelope): Envelope;

    // ReceiverInterface
    public function get(): iterable; // retourne les Envelopes à traiter
    public function ack(Envelope $envelope): void;    // succès
    public function reject(Envelope $envelope): void; // échec définitif
}

Transports fournis

Configuration YAML

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    multiplier: 2
            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            App\Message\SendEmailCommand: async
            'App\Message\*': async # wildcard

#[AsMessage] — routing inline sur la classe

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage(transport: 'async')]
final readonly class SendEmailCommand
{
    public function __construct(public readonly string $to) {}
}

Middleware — la chaîne de responsabilité

Le bus est une chaîne de middlewares — chaque message les traverse dans l'ordre avant d'atteindre les handlers.

Le bus Messenger est une chaîne de middlewares — chaque message la traverse de bout en bout. Les middlewares peuvent lire l'Envelope, l'enrichir avec des stamps, court-circuiter la chaîne, ou déléguer au suivant. C'est le pattern Chain of Responsibility appliqué au bus.

MiddlewareInterface

interface MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope;
}

Pour passer au middleware suivant : return $stack->next()->handle($envelope, $stack);. Pour court-circuiter : retourner directement une Envelope sans appeler $stack->next().

Middleware custom — exemple

use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;

final class LoggingMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $class = $envelope->getMessage()::class;
        $this->logger->info("Dispatch: {$class}");

        // Passer au middleware suivant et récupérer l'Envelope enrichie
        $envelope = $stack->next()->handle($envelope, $stack);

        $handled = $envelope->last(HandledStamp::class) !== null;
        $this->logger->info("Done: {$class}", ['handled' => $handled]);

        return $envelope;
    }
}

Les middlewares intégrés — ordre par défaut

Visualise la chaîne ci-dessous :

Worker — consommer les messages

Le Worker est le processus qui boucle sur les transports et dispatch chaque message reçu au bus.

Le Worker est un processus de longue durée qui boucle sur un ou plusieurs transports, récupère les messages et les dispatche au bus. C'est le moteur qui alimente le traitement asynchrone.

Cycle de vie d'un message dans le Worker

Le Worker appelle $transport->get() en boucle. Pour chaque message reçu, il ajoute un ReceivedStamp (avec le nom du transport) et dispatche l'Envelope dans le bus — ce qui déclenche les middlewares et le ou les handlers.

Worker::run() — les options

use Symfony\Component\Messenger\Worker;

$worker->run([
    'sleep'       => 200000, // µs entre deux polls si aucun message (défaut: 200000 = 0.2s)
    'timeLimit'   => 3600,   // secondes — s'arrête après N secondes
    'memoryLimit' => 128e6,  // octets — s'arrête si la mémoire dépasse ce seuil
    'limit'       => 100,    // nombre de messages — s'arrête après N messages traités
]);

Commande console

# Consommer depuis le transport "async"
php bin/console messenger:consume async

# Avec des limites
php bin/console messenger:consume async --time-limit=3600 --memory-limit='128M'

# Consommer depuis plusieurs transports
php bin/console messenger:consume async failed

ack() et reject()

Après le dispatch par le Worker :

• Si le handler réussit → $transport->ack($envelope) — le message est supprimé de la file
• Si une exception non-récupérable est levée → $transport->reject($envelope) — le message est supprimé sans retry
• Si une exception récupérable est levée et que des retries sont configurés → le message est renvoyé avec un RedeliveryStamp

Arrêt propre

Le Worker écoute les signaux POSIX. Sur SIGTERM ou SIGINT, Worker::stop() est appelé : le Worker finit le message en cours puis s'arrête proprement — aucun message perdu.

Retry et gestion des erreurs

Quand un handler échoue, Messenger peut réessayer automatiquement — avec délai exponentiel et transport d'échec.

Quand un handler lève une exception, Messenger ne jette pas le message à la poubelle. Il applique une stratégie de retry : attendre un délai croissant, puis réessayer — jusqu'à atteindre le nombre maximal de tentatives, après quoi le message part dans le transport failed.

RetryStrategyInterface

interface RetryStrategyInterface
{
    public function isRetryable(Envelope $message, ?\Throwable $throwable = null): bool;
    public function getWaitingTime(Envelope $message, ?\Throwable $throwable = null): int; // en ms
}

MultiplierRetryStrategy

use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy;

new MultiplierRetryStrategy(
    maxRetries:           3,
    delayMilliseconds:    1000,  // délai initial en ms
    multiplier:           2,     // chaque retry multiplie le délai précédent
    maxDelayMilliseconds: 0,     // 0 = pas de plafond
    jitter:               0.1,   // fraction aléatoire (évite les tempêtes de retry)
)

Avec maxRetries=3, delay=1000, multiplier=2 :

• Tentative 1 : échec → attente ~1000 ms
• Tentative 2 : échec → attente ~2000 ms
• Tentative 3 : échec → attente ~4000 ms
• Tentative 4 : échec → transport failed (plus de retry)

Contrôler le retry depuis le code

use Symfony\Component\Messenger\Exception\RecoverableExceptionInterface;
use Symfony\Component\Messenger\Exception\UnrecoverableExceptionInterface;

// Exception récupérable → retry autorisé même si isRetryable() retournerait false
final class ApiRateLimitException extends \RuntimeException
    implements RecoverableExceptionInterface {}

// Exception non-récupérable → envoi direct au failed transport, pas de retry
final class InvalidPayloadException extends \RuntimeException
    implements UnrecoverableExceptionInterface {}

Lance la simulation ci-dessous pour voir les 3 tentatives avec délais croissants :

DispatchAfterCurrentBusStamp

Ce stamp permet de différer l'exécution d'un message jusqu'à ce que le bus courant ait fini — pour garantir la cohérence des transactions.

Imagine un handler qui crée un utilisateur en base, puis dispatche un événement UserCreatedEvent. Si un listener de cet événement lit la base immédiatement, il risque de ne pas trouver l'utilisateur — la transaction n'est peut-être pas encore committée. DispatchAfterCurrentBusStamp résout ce problème.

Fonctionnement

use Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp;

// Dans un handler :
public function __invoke(CreateUserCommand $cmd): void
{
    $user = User::create($cmd->email);
    $this->em->persist($user);
    $this->em->flush(); // commit de la transaction

    // Cet événement sera dispatché APRÈS que le bus courant ait terminé
    $this->bus->dispatch(
        new UserCreatedEvent($user->getId()),
        [new DispatchAfterCurrentBusStamp()]
    );
}

Séquence d'exécution

1. Le handler dispatche UserCreatedEvent avec le stamp
2. DispatchAfterCurrentBusMiddleware intercepte le message et le met en file d'attente interne
3. Le bus courant termine (handler de CreateUserCommand retourne)
4. Le middleware dispatche tous les messages en attente dans l'ordre

Multiple buses et ValidationStamp

Symfony permet plusieurs buses avec des responsabilités séparées — command bus, query bus, event bus — chacun avec ses middlewares.

Une application complexe peut avoir plusieurs buses avec des responsabilités bien séparées : un command bus pour les mutations, un query bus pour les lectures, un event bus pour les notifications. Chacun peut avoir ses propres middlewares.

Configuration de plusieurs buses

# config/packages/messenger.yaml
framework:
    messenger:
        default_bus: command.bus

        buses:
            command.bus:
                middleware:
                    - validation
                    - doctrine_transaction

            query.bus:
                middleware: []  // pas de transaction pour les queries

            event.bus:
                default_middleware: allow_no_handlers  // ok si personne n'écoute

Restreindre un handler à un bus

#[AsMessageHandler(bus: 'command.bus')]
final class CreateUserHandler { ... }

#[AsMessageHandler(bus: 'query.bus')]
final class FindUserHandler { ... }

ValidationMiddleware et ValidationStamp

Le ValidationMiddleware n'est actif que si ValidationStamp est présent dans l'Envelope. Ajoute-le au dispatch pour déclencher la validation Symfony :

use Symfony\Component\Messenger\Stamp\ValidationStamp;
use Symfony\Component\Validator\Constraints as Assert;

// Sur la classe du message :
final readonly class CreateUserCommand
{
    public function __construct(
        #[Assert\Email]
        public readonly string $email,
    ) {}
}

// Au dispatch :
$bus->dispatch(
    new CreateUserCommand($email),
    [new ValidationStamp([Default::class])]
);

Si la validation échoue → ValidationFailedException est levée avant d'atteindre les handlers.

Injecter un bus spécifique

use Symfony\Component\DependencyInjection\Attribute\Target;

public function __construct(
    #[Target('command.bus')]
    private readonly MessageBusInterface $bus,
) {}

# services.yaml
App\Service\MyService:
    bind:
        MessageBusInterface $bus: '@command.bus'

Synthèse — le guide de choix

Messenger ou EventDispatcher ? Async ou sync ? Le guide de choix complet.

Messenger est un outil puissant — mais ce n'est pas toujours le bon. Voici les clés pour choisir entre Messenger et EventDispatcher, entre sync et async, entre les transports.

Messenger vs EventDispatcher

Guide de choix du transport

Anti-patterns à éviter

Récapitulatif des stamps

Récapitulatif des middlewares intégrés