Utiliser le Messenger de Symfony avec RabbitMq

Introduction

Dans le cadre de certains de nos projets, nous devons récupérer des payloads JSON venu d’autre application / langage via des queues RabbitMQ (protocole AMQP), cependant le Messenger de Symfony utilise une enveloppe spécifique pour transmettre ses messages, il nous faut donc adapter la réception des messages dans le Messenger pour utiliser un simple payload json (sans Enveloppe).

Installation de Symfony et des composants

Tout d’abord nous allons créer le projet Symfony :

composer create-project/skeleleton rabbitmq_worker

Ensuite nous importons le composant Doctrine :

composer require symfony/orm-pack

Puis importer le composant “Messenger” de Symfony qui nous permettra de consommer nos messages du RabbitMq :

composer require symfony/messenger

Vous devez aussi ajouter la librairie « amqp-lib », qui nous permettra de communiquer avec RabbitMq :

composer require php-amqplib/php-amqplib

Le code

Nous allons avoir besoin de 3 classes :

App\Message\ImageUploaded: L’objet qui représente notre « type » de message
App\Messenger\ExternalImageUploadedSerializer : Cette class va décoder le payload json pour créer l’enveloppe du Messenger
App\MessageHandler\MessageHandler : Notre handler qui va réceptionner et traiter le message

App\Message\ImageUploaded.php

<?php
namespace App\Message;

/**
* Class ImageUploaded
*/ 
class ImageUploaded
{
    /** 
     * @var int
     */
    private $id;

    /** 
     * @var string
     */
    private $path;

    /**
     * ImageUploadedconstructor.
     * @param string $message
     */
    public function __construct(int $id, string $path)
    {
        $this->id = $id;
        $this->path= $path;
    }

    /**
     * @return int
     */
    public function getId(): int
    {
        return $this->id;
    }

    /**
     * @return string
     */
    public function getPath(): string
    {
        return $this->path;
    }
}

App\Messenger\ExternalImageUploadedSerializer.php

<?php
namespace App\Messenger;

use App\Message\ImageUploaded;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;

/**
* Class ExternalImageUploadedSerializer 
*/
class ExternalImageUploadedSerializer implements SerializerInterface
{
    public function decode(array $encodedEnvelope): Envelope
    {
        $body = $encodedEnvelope['body'];
        $data = json_decode($body);
        // TODO: check message integrity (id, and path)
        $message = new ImageUploaded($data['id'], $data['path']);
        $stamps = [];

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

    public function encode(Envelope $envelope): array
    {
        return [];
    }
}

App\MessageHandler\ImageUploadedHandler.php

<?php
namespace App\MessageHandler;

use App\Message\ImageUploaded;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
use Symfony\Component\Serializer\SerializerInterface;

/*
 * Class ImageUploadedHandler 
 */
class ImageUploadedHandler implements MessageHandlerInterface
{
    /*
     * @var SerializerInterface
     */
    private EntityManagerInterface $entityManager;

    /**
     * class constructor
     *
     * @param EntityManagerInterface $entityManager
     */
    public function __construct(EntityManagerInterface $entityManager,MessageRepository $messageRepository)
    {
        $this->entityManager = $entityManager;
    }

    /**
     * @param ImageUploaded $imageUploaded
     */
    public function __invoke(ImageUploaded $imageUploaded)
    {

        $imageId = $imageUploaded->getId();
        $path = $imageUploaded->getPath();
        
        // resize
        // crop
        // update storage / database
        // send other message in queue (end job)
        // ...
    }
}

Configuration du Messenger

Nous devons ensuite configurer le Messenger pour qu’il puisse router correctement le message dans notre handler.

// config/packages/messenger.yaml

framework:
    messenger:
        serializer:
          default_serializer: messenger.transport.symfony_serializer
          symfony_serializer:
            format: json
            context: { }
  

        transports:
            image_uploaded:
              dsn: '%env(RABBITMQ_DSN)%'
              serializer:  App\Messenger\ExternalImageUploadedSerializer
              options:
                queue_name: image_uploaded
    
        routing:
             'App\Message\ImageUploaded': image_uploaded

Traiter les messages

Vous pouvez maintenant lancer la commande suivante, pour traiter les messages envoyés dans RabbitMQ

php bin/console messenger:consume image_uploaded

Conclusion

Nous utilisons ici qu’un petit pourcentage des possibilités du Messenger de Symfony, nous ne manquerons pas de publier de nouveaux articles sur ce sujet !