Symfony Messenger et l’interopérabilité

Le composant Messenger a été mergé dans Symfony 4.1, sorti en mai 2018. Il ajoute une couche d’abstraction entre un producteur de données (publisher) et son consommateur de données (consumer).

Symfony est ainsi capable d’envoyer des messages (la donnée) dans un bus, le plus souvent asynchrone. Concrètement : notre controller crée un email, l’envoie dans un bus (RabbitMQ, Doctrine, …) et un consommateur envoie l’email de manière synchrone. L’avantage est de déléguer les tâches lourdes, longues, ou sensibles à des workers en arrière-plan.

Quand le producteur et le consommateur de la donnée sont dans la même application, ce système fonctionne très bien et de manière totalement transparente. Cependant, si ce sont deux applications Symfony différentes, ou deux applications dans deux langages différents, il va falloir préparer un peu le terrain.

Section intitulée l-architecture-de-messengerL’architecture de messenger

L’architecture globale du composant est la suivante :

1. Un publisher (controller, service, command, …) dispatche un message dans le bus ;
2. Si le bus est synchrone, le message est consommé par un handler ;
3. Si le bus est asynchrone, le message est envoyé via un transport à un système de queue (RabbitMQ, Doctrine, …) ;
4. Un daemon (aussi appelé worker) va chercher en temps réel les messages depuis le système de queue via le transport ;
5. Il re-dispatche le message dans le bus ;
6. Maintenant le bus est synchrone, le message est consommé par un handler.

Quand un bus est synchrone, tout se passe naturellement. Cependant si le transport est asynchrone, alors votre message doit être sérialisé. Symfony supporte nativement deux modes de sérialisation :

• La sérialisation native de PHP ;
• La sérialisation via le composant Serializer.

Section intitulée la-serialisation-de-messageLa sérialisation de message

Par défaut, un message est sérialisé avec PHP et il ressemble à ça :

O:36:\"Symfony\\Component\\Messenger\\Envelope\":2:{s:44:\"\0Symfony\\Component\\Messenger\\Envelope\0stamps\";a:1:{s:46:\"Symfony\\Component\\Messenger\\Stamp\\BusNameStamp\";a:1:{i:0;O:46:\"Symfony\\Component\\Messenger\\Stamp\\BusNameStamp\":1:{s:55:\"\0Symfony\\Component\\Messenger\\Stamp\\BusNameStamp\0busName\";s:21:\"messenger.bus.default\";}}}s:45:\"\0Symfony\\Component\\Messenger\\Envelope\0message\";O:18:\"App\\Message\\Foobar\":1:{s:24:\"\0App\\Message\\Foobar\0name\";s:6:\"coucou\";}}

Nous pouvons voir App\\Message\\Foobar, ce qui représente la classe de l’objet contenu dans le message.

D’une application à une autre, cette classe peut ne pas exister. Il y a même très peu de chance qu’elle existe. Et si l’autre application est dans un autre langage, il est impossible (ou presque 😈) de désérialiser du PHP !

Nous allons utiliser un format d’échange plus classique. Nous avons choisi le JSON, mais nous aurions pu utiliser le XML, Protobuf, ou n’importe quel langage de sérialisation interopérable.

Section intitulée un-serialiseur-sur-mesureUn sérialiseur sur mesure

Il va nous falloir implémenter l’interface suivante SerializerInterface du composant :

namespace Symfony\Component\Messenger\Transport\Serialization;

use Symfony\Component\Messenger\Envelope;

interface SerializerInterface
{
    public function decode(array $encodedEnvelope): Envelope;

    public function encode(Envelope $envelope): array;
}

• La méthode decode() désérialise ce qui vient de notre transport ;
• La méthode encode() sérialise notre objet métier pour le transport.

Afin de ne faire qu’un sérialiseur pour tous les messages qui transitent entre nos deux applications, nous allons utiliser une clé type qui permettra d’identifier quelle classe / donnée nous sommes en train de manipuler. Nous allons créer une interface pour nos messages :

namespace App\Messenger\Serializer;

interface JsonSerializableInterface
{
    public function getJsonType(): string;
}

Prenons l’exemple d’un objet Foobar :

final class Foobar implements JsonSerializableInterface
{
    public function __construct(
        public readonly string $name,
    ) {
    }

    public function getJsonType(): string
    {
        return 'foobar';
    }
}

Une fois sérialisé, il ressemblera à ça :

{"name":"coucou"}

Et voici le code de notre sérialiseur :

namespace App\Messenger\Serializer;

use App\Messenger\Foobar;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;

class JsonSerializer implements SerializerInterface
{
    public function decode(array $encodedEnvelope): Envelope
    {
        $message = match ($encodedEnvelope['headers']['type']) {
            Foobar::class => new Foobar($encodedEnvelope['body']['name']),
            default => throw new \LogicException('The message type is not supported.'),
        };

        $stamps = [];
        $retryCount = 0;
        foreach ($encodedEnvelope['headers']['x-death'] ?? [] as ['count' => $count]) {
            $retryCount += $count;
        }
        if ($retryCount) {
            $stamps[] = new RedeliveryStamp($retryCount);
        }

        return new Envelope($message, $stamps);
    }

    public function encode(Envelope $envelope): array
    {
        $message = $envelope->getMessage();

        if (!$message instanceof JsonSerializableInterface) {
            throw new \LogicException(sprintf('The message must implement "%s".', JsonSerializableInterface::class));
        }

        return [
            'body' => json_encode($message),
            'headers' => [
                'type' => $message->getJsonType(),
                'Content-Type' => 'application/json',
            ],
        ];
    }
}

Et voilà ! Il ne reste plus qu’à brancher ce sérialiseur dans la configuration de notre transport :

framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                serializer: App\Messenger\Serializer\JsonSerializer

Section intitulée conclusionConclusion

Faire communiquer deux applications PHP ou non via un bus de données est quelque chose de simple à faire avec Symfony. Comme souvent, en implémentant une interface et avec un peu de configuration, nous remplaçons une partie de Symfony pour l’adapter à notre besoin.

Il est possible d’améliorer le code que nous vous avons proposé. Il faudrait mieux valider la donnée qui rentre dans l’application (ie: valider le $body). Nous pourrions aussi utiliser le Serializer de Symfony pour faire la conversion « nos objets PHP » <-> JSON, mais ce n’est pas le but de cet article. Vous pouvez utiliser par exemple Happyr/message-serializer.

Soyez créatifs, faites de votre mieux, et si possible du bon boulot. Mais n’oubliez pas : Faites du code simple !

Au passage, j’ai été amené à faire ça dans le cadre d’un projet personnel IoT. Mes microcontrôleurs envoient de la donnée via MQTT (serveur RabbitMQ) et une application Symfony consomme cette donnée pour la publier dans InfluxDB. Peut-être un prochain article en perspective 😍