Symfony Messenger : tache en arrière plan

Bonjour,

Je souhaiterais vous montrer comment créer une tâche en arrière plan en utilisant le composant Messenger du framework Symfony.

Vous avez un gros script à exécuter (import / export de fichier par exemple) depuis votre application web, tout en voulant garantir un traitement efficient : C'est un cas de figure ou le composant Messenger peut vous être très utile.

Pour l'exemple, imaginons que nous disposons d'un service AwesomeService qui permet d'exécuter une lourde méthode d'import en données en base.

<?php
#src\Service\AwesomeService.php

namespace App\Service;

use Doctrine\ORM\EntityManagerInterface;

class AwesomeService
{
    private $em;

    public function __construct(EntityManagerInterface $em) 
    {
        $this->em = $em;
    }

    public function import(array $parameters): array
    {
        // Cette méthode est pour illustrer l'exemple
        $results = [];

        // Faire quelque chose, comme insérer en masse en base

        return $results;
    }
}

Installation du composant Messenger

Installer le composant via composer :

php composer require symfony/messenger

[Recommandé] Cet autre composant vous permettra de stocker les informations des messages dans la table messenger_messages de votre base de données :

php composer require symfony/doctrine-messenger

Ce qui pourra être très utile pour la gestion des essais / échecs des scripts, pour les relancer via une autre queue de transport.

Création d'un message et de son gestionnaire

Ce composant fonctionne d'une manière analogue à celui évoqué dans mon précédent article sur les écouteurs d'évènements.

Créons tout d'abord un message :

<?php
# src\Messenger\Message\AwesomeMessage.php

namespace App\Messenger\Message;

class AwesomeMessage
{
    private $userId;
    private $parameters;

    public function __construct(int $userId, ?array $parameters = []) 
    {
        // Nous pourrons récupérer l'utilisateur en cours, grâce au $userId
    }

    public function getUserId(): int
    {
        return $this->userId;
    }

    public function getParameters(): array
    {
        return $this->parameters;
    }
}

Créons maintenant le gestionnaire correspondant, permettant d'exécuter la lourde méthode du service et au choix, d'envoyer un e-mail ou un autre message

<?php
# src\Messenger\MessageHandler\AwesomeMessageHandler.php

namespace App\Messenger\MessageHandler;

use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
use Doctrine\ORM\EntityManagerInterface;

// Pour envoyer un mail
use Symfony\Component\Mailer\MailerInterface;
use Symfony\Component\Mime\Email;
use Symfony\Component\Mailer\Exception\TransportExceptionInterface;

// Pour envoyer un autre message
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp;

use App\Messenger\Message\AwesomeMessage;
use App\Service\AwesomeService;

class AwesomeMessageHandler implements MessageHandlerInterface
{
    private $awesome_service;
    private $em;
    private $message_bus;
    private $mailer;

    public function __construct( 
        AwesomeService $awesome_service, 
        EntityManagerInterface $em,
        MessageBusInterface $message_bus,
        MailerInterface $mailer
    ) {
        $this->awesome_service = $awesome_service;
        $this->em = $em;
        $this->message_bus = $message_bus;
        $this->mailer = $mailer;
    }

    public function __invoke(AwesomeMessage $message)
    {
        $user = $this->em->getRepository(User::class)->findOneById($message->getUserId());
        $results = $this->$awesome_service->import($message->getParameters());

        // Par exemple, si erreur, envoyer un mail
        if (true === $results['has_error'])
        {
            $email = (new Email())
                ->from('hello@example.com')
                ->to($user->getEmail())
                ->subject('Time for Symfony Mailer!')
                ->text('Sending emails is fun again!')
                ->html('<p>See Twig integration for better HTML integration!</p>');

            try {
                $this->mailer->send($email);
            } catch (TransportExceptionInterface $e) {

            }
        }

        // On pourrait aussi dispatcher un autre message
        $new_message = new AwesomeOtherMessage($results); // A Créer
        $this->message_bus->dispatch(
            (new Envelope($new_message))
                ->with(new DispatchAfterCurrentBusStamp())
        );
    }
}

Utiliser le composant Messenger

Nous allons maintenant créer un controlleur qui dispatchera notre message

<?php
// src/Controller/DefaultController.php
namespace App\Controller;

use App\Messenger\Message\AwesomeMessage;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Messenger\MessageBusInterface;

class DefaultController extends AbstractController
{
    public function index(MessageBusInterface $bus): Response
    {
        // Le AwesomeMessageHandler sera appellé avec en paramètre l'identifiant de l'utilisateur courant
        $bus->dispatch(new AwesomeMessage($this->getUser()->getId()));
        // ...
    }
}

Configurer le composant Messenger

Créons désormais le fichier de configuration du composant

# config/packages/messenger.yaml
framework:
    messenger:
        # reset services after consuming messages
        reset_on_message: true
        failure_transport: failed

        transports:
            # https://symfony.com/doc/current/messenger.html#transport-configuration
            async: '%env(MESSENGER_TRANSPORT_DSN)%'
            failed: '%env(MESSENGER_TRANSPORT_DSN)%?queue_name=failed' # Les messages en erreur seront stockés dans la table de votre base de données
            sync: 'sync://'

        routing:
            # Route your messages to the transports
            'App\Messenger\Message\AwesomeMessage': sync # Synchrone : Pour tests
            'App\Messenger\Message\AwesomeMessage': async # En arrière plan

Tester l'acheminement des messages via la ligne de commande

Le composant comprend des commandes qui permettent d'acheminer les message depuis une certaine queue de transport (Ici async)

php bin/console messenger:consume async -vv 

Déploiement en production

Il faut créer un service autonome qui aura pour tâche de traiter tous les messages

[Unit]
Description=Symfony messenger-consume %i

[Service]
ExecStart=php /path/to/your/app/bin/console messenger:consume async --time-limit=3600
Restart=always
RestartSec=30

[Install]
WantedBy=default.target

Activation du service

sudo systemctl enable messenger-worker@service
sudo systemctl start messenger-worker@service

Pour plus d'informations, voir la documentation du Composant Messenger (En anglais)