Nous présentons aujourd'hui un petit retour d'experience sur le composent Symfony Messenger. L'objectif derrière ce billet est de comprendre comment:

1. Mettre en place une architecture de conteneur Docker qui repond à notre besoin.
2. Fonctionnement de l'ordonnenacement de RabbitMQ.
3. Fonctionnement de Symfony Messenger.
4. Le couplage de Symfony Messenger et RabbitMQ pour envoyer des email d'une facon asynchrone.

De nombreux de traitement web pourait etre assez lourd à cause et prend parfois bcp de temps parfois à cause trafic, à cause de l'architecture logiciel mise en place, à cause du gestion de mémoire de notre serveur, ya des gens qui perfère augmenter la mpémoire ou bien mettre un système de balance loader mais ca résoud pas le problème à long terme et c'est payant .

Bref, pour résoudre ce problème, on fait retour souvent à un mécanisme de file d'attente (ex: Standard Queue, FIFO Queue) pour effectuer des tâches chronophages et d’une manière ordoannacée et plus performante.

RabbitMQ est un système permettant de gérer des files de messages affin de permettre à différents clients de communiquer très simplement et plus efficassement. Pour que chaque client puisse communiquer avec RabbitMQ, celui-ci s’appuie sur le protocole AMQP. Ce protocole définit précisément la façon d’ordonnacement asynchronne des messages. C’est un gestionnaire de queues, permettant d’asynchroniser différents traitements. C’est pour assurer la haute disponibilité du serveur, dédié au d’exécution de traitements lourds comme l’envoi des e-mails, l’exportation, l'importation des fichiers et d'autres traitements durable.

Le composant messenger permet à l'application d'envoyer et de recevoir des messages vers / depuis d'autres applications ou via un système de queue basé sur un message bus. Un message bus est un composant très simple qui permet de dispatcher un objet (message). Le bus a pour rôle d’exécuter le handler approprié au message disant 'déspatché'. Durant ce processus le message va passer dans une pile ordonnée de notre middleware.

Le composant nous fournit aussi des fonctionnalités de routage pour intercepter les messages et les router vers le bon Transporter déja définit au niveau de notre configuration config.yml.

Bref, notre nouveau composant est composé par(single responsabilité poo solid) :

Message : Objet PHP qui doit être sérialisable. Data transfer object (DTO).

Transporter: Par défaut, les messages sont traités dès leur distribution. Si vous préférez traiter les messages de manière asynchrone, vous devez configurer un transport. Ces transports communiquent avec notre application via des systèmes de file d'attente ce qu'on applleles brocker(RabbitMQ, Kafka, AWS SQS, etc.).

Message Handler : La classe NotificationHandler qui recevra les messages. C’est ici qu’il pourra y avoir et executer la logique métier applicable aux messages.

Sender (Expéditeur) : Responsable de la sérialisation et de l'envoi de messages à quelque chose . Cela peut être un courtier de messages ou une API tierce par exemple.

Receiver (Récepteur) : Responsable de la désérialisation et de la transmission des messages aux gestionnaires. Cela peut être un extracteur de file d'attente de messages ou un point de terminaison d'API, par exemple.

Message Bus : Définir le handler coresspondant associer à notre message. Le bus est utilisé pour envoyer des messages. Un message bus est un composant très simple qui permet de dispatcher un objet (message). Le bus a pour rôle d’exécuter le handler approprié au message, Le cœur du composant.

D'ailleurs le Message Bus represente une approche de developpement suivit par DDD (developpement dérigé par doamin) qui présente une architecture d'event sourcing ce q'on appelle CQRS. L'un des architecture les plus recommendé qui résoudre énormement de problème de perfermance et de comlexité algorithmique.

Worker: Consomme les messages depuis les transports.

Pourquoi utiliser un transport?
Tronspoter : ils permettent de faire transiter les messages via différents logiciels (RabbitMQ, Kafka, Amazon SQS, etc.) 

A la place d'envoyer notre message d'une manière synchrone directement vers notre Handler, on dérige le message vers un transporter pour le fournir d'une facon asynchrone à notre message Handler. C'est Cool !

Le composant bénéficie d’un panel sur la toolbar de debug, mais aussi d’une commande afin de pouvoir consommer les messages à travers la commande suivante: bin/console messenger:consume-messages 'nom de Tronspoter(amqp, default)'

Ok on y va !! … Nous allons présenter comment utiliser le compossant Messenger de Symfony pour pouvoir mettre en place un système d'envoi de mail d'un facon asynchrone. Imganions si nous devons enovyer à 60000 utilisateurs un mail de notification, la page correspendante à notre route va continuer à charger tantque le boucle for ($i = 1; $i <= 6000; $i++) n'a pas terminé son parcours, et du coup ca va consommer énormement de mémoire et de plus ce n'est pas pratique ni supportable en terme d'User Experience.

Pour se faire, nous avons deux choix soit d'installer notre environnement directement sur notre machine local soit utiliser un système de conteneur:
La 1ere Méthode:

rabbitmqctl
1. Installer rabbitmq et lancer la commande rabbitmq-server 
2. rabbitmqctl status
3. rabbitmq-plugins enable rabbitmq_management
4. afin d'activer l'extension amqp. Il faudrait lancer 'apt install php-amqp' et ajouter 'extension = amqp.so' sous le fichier php.ini

5.composer create-project symfony/skeleton my-project # c'est mieux de travailler avec le projet symfony/skeleton plus lègère avec le Minimum de dépenses possibles et installées à fur et aux mesures les composants nécessaires.

6. La mise en place des dépendences necessaires:

composer req mailer
composer messenger
composer req annotation
composer req serializer
composer req twig
composer req web-server --dev

La 2eme Méthode: On prépare notre docker-compose.yml pour regroupper les différents configurations:

Voici mon docker-compose.yml:

version: '2'

services:
    php:
        build: docker/php
        tty: true
        restart: on-failure
        volumes:
          - '.:/symfony'
        command: service supervisor start
        links:
            - rabbitmq

    nginx:
         build: docker/nginx
         tty: true
         restart: on-failure
         volumes_from:
           - php
         ports:
             - '80:80'
         links:
             - php

    rabbitmq:
        image: rabbitmq:3.4-management
        tty: true
        ports:
              - "15672:15672"

    maildev:
        image: djfarrelly/maildev
        tty: true
        ports:
              - "1080:80"

La création de notre Message : SendNotification

namespace App\Message;

class SendNotification
{
    private $message;

    private $users;

    public function __construct(string $message, array $users = [])
    {
        $this->message = $message;
        $this->users = $users;

    }
...

Voici notre controlleur:

Nous allons utlisé notre command Bus pour dispatcher notre command, autrement dit notre message de type SendNotification pour que le Handler puisse sera notifier de cette action et répondre à travers le traitement necessaire.

namespace App\Controller;

class HomeController extends AbstractController
{
    /**
     * @Route("/", name="home")
     */
    public function index(MessageBusInterface $bus)
    {
        $bus->dispatch(new SendNotification('notification'));

        return $this->render('home/index.html.twig');
    }
...

La création de notre Handler: (par convention faudrait que le nom de la callse soit 'nom de la classe de notre message'+Handler. Pour traiter un message il faudra créer un handler avec une méthode __invoke (si vous avez déjà une méthode, utilisez l’attribut handles dans le tag de définition du service.

namespace App\MessageHandler;

class SendNotificationHandler
{
    /**
     * @var Swift_Mailer
     */
    private $mailer;

    public function __construct(Swift_Mailer $mailer)
    {
        $this->mailer = $mailer;
    }

    public function __invoke(SendNotification $notification)
    {
        for ($i =1; $i<5000000; $i++) {
            echo 'The mail has been sent with success !';
            $message = (new \Swift_Message('Mail object'))
                ->setFrom('[email protected]')
                ->setTo('test@gmailcom')
                ->setBody(
                    $notification->getMessage()
                )
            ;

            $this->mailer->send($message);
        }
    }
}
...

Ensuite, il faudrait enregistrer notre Handler en tant que service avec le tag messenger.message_handler:

# config/services.yaml
services:
    App\MessageHandler\:
        resource: '../src/MessageHandler'
        tags: [messenger.message_handler]

On pourait aussi définir notre service avec le syntaxe complète afin de spécifier la méthode à appeler et le modèle supporté.

# config/services.yaml
services:
    App\MessageHandler\NotificationHandler:
        tags:
             - {name: 'messenger.message_handler', method: 'process', handles: 'App\Message\SendNotification'} # telque 'process' reprèsente est une méthode deja définit au niveau de la classe SendNotificationHandler

Transporteur/sender RabbitMQ:

Au lieu d’appeler directement notre Handler, nous voulons bien acheminer les messages vers un ou plusieurs expéditeurs/transports:

framework:
    messenger:
        transports:
            # Uncomment the following line to enable a transport named "amqp"
             amqp: '%env(MESSENGER_TRANSPORT_DSN)%'

        routing:
            # Route your messages to the transports
             'App\Message\SendNotification': amqp

.nev : Ici c'est l'endroit ou définit notre variable denvironnement MESSENGER_TRANSPORT_DSN

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

Une fois que les messages ont été acheminés, nous pouvons les consommer avec la commande $ bin/console messenger:consume-messages messenger.default_receiver

et s'intervient Supervisor pour executer la commande d'une facon automatique, faudrait prendre en consédiration que le lancememnt de service supervisor, arrète le service php-fpm, c'est pour cela que je viens d'indiquer au niveau de la configuration la relancememnt de php-fpm en cas d'arret.

[unix_http_server]
file=/tmp/supervisor.sock
[supervisord]
logfile=/tmp/supervisord.log
pidfile=/var/run/supervisord.pid
nodaemon=true
[program:php-fpm]
command = /usr/local/sbin/php-fpm
autostart=true
autorestart=true

[program:amqp]
command=php /symfony/bin/console messenger:consume-messages amqp
startsecs = 0
stdout_logfile=/tmp/supervisord-amqp.log
stdout_logfile_maxbytes=10MB

On pourait aussi lancer la commande bin\console debug:messenger c'est une commande CLI pour exposer les classes de message pouvant être dispatché à travers notre Messenger.

#config.yml
swiftmailer:
    url: '%env(MAILER_URL)%'
  #  spool: ~

#.env
MAILER_URL=smtp://maildev:25

Nous avons commenter le mail spool parceque nous utilisons déja un transporteur amqp pour enchainer l'envoi d'une facon asynchrone donc ya pas besoin de traiter le processus avec un système de spooling. (c'est la solution la plus rapide)

Pour pouvoi utiliser d'autre transporteurs nous aloons installer req enqueue/messenger-adapter qui va nous fournir ce qu'on appelle un adapter

tout d'abord faudrait creer le service à partir de SQS Managment Console et récupérer le nom la région et les clés d'authentification qui se trouvent sous l'interface IAM.

Après il suffit d'installer les dépedences necessaires: composer require messenger enqueue/messenger-adapter enqueue/sqs pour pouvoir utiliser d'autres type de transports nous avons fait recours à ce package https://github.com/adtechpotok/enqueue-messenger-adapter

Voici les différents configurations necessaires :

.env
ENQUEUE_DSN=sqs:?secret=[AWSSecretKey]&key=[AWSAccessKey]region=[region]

messenger.yml

framework:
    messenger:
        transports:
            # Uncomment the following line to enable a transport named "amqp"
             sqs: enqueue://default?&topic[name]=test&queue[name]=test

        routing:
            # Route your messages to the transports
             'App\Message\SendNotification': sqs

Afin de traiter le message, il suffit de lancer la commande: php bin/console messenger:consume sqs

Nous pouvons visualiser, lire le contnue, purger le queue à travers l'interface offert par AWS SQS: