Découvrir les nouveautés du composant Messenger avec Symfony 7.2

Le composant Messenger permet d’envoyer et de traiter des messages de manière synchrone ou asynchrone via des transports (queues). Les éléments principaux de Messenger sont les messages (une classe PHP qui contient des données), les handlers (une classe PHP appelée lorsque des messages qu’elle supporte sont dispatchés) et les transports (qui gèrent l’envoi et la distribution des messages). 

Si vous n’êtes pas familier avec le composant, n’hésitez pas à consulter la documentation officielle.

Le composant Messenger a évolué en deux points dans la version 7.2 de Symfony : 

• L’arrivée d’un nouvel attribut [#AsMessage]
• La mise en place d’une connexion KeepAlive pour éviter les traitements de messages superflus 

L’attribut [#AsMessage] 

Pierre Rineau a contribué à ces évolutions, en ajoutant de l’attribut #[AsMessage], permettant de configurer les transports directement au sein des classes de message.

Par exemple, si vous avez une classe de message comme celle-ci : 

 // src/Message/SomeMessage.php 
namespace App\Message; 
 
class SomeMessage
{ 
   public function __construct( 
       private string $content, 
   )
   { 
   } 
 
   public function getContent(): string 
   { 
       return $this->content; 
   } 
} 

En supposant que vous ayez également défini une classe SomeMessageHandler dans votre application, vous pouvez utiliser la configuration suivante pour indiquer à Symfony que le message doit être envoyé à un transport/queue appelé async : 

 # config/packages/messenger.yaml 
framework: 
   messenger: 
       transports: 
           async: "%env(MESSENGER_TRANSPORT_DSN)%" 
 
       routing: 
           'App\Message\SomeMessage': async 

Ce nouvel attribut vous permet de configurer le ou les transports directement dans la classe du message. Dans l’exemple ci-dessus, vous pouvez maintenant supprimer la configuration « routing: » dans messenger.yaml et ajouter ce qui suit à la classe du message : 

#[AsMessage('async')] 
class SomeMessage
{ 
   // ... 
} 

Ici un exemple avec des transports multiples :

#[AsMessage(['async', 'audit'])] 
class SomeMessage
{
    // ....
}
 

⚠️ Point de vigilance : Si vous définissez à la fois une configuration YAML/XML et un attribut PHP, la configuration dans Symfony prend toujours le dessus sur l’attribut de la classe. Ce comportement permet de surcharger le routage en fonction de l’environnement.

Veillez donc a supprimer votre configuration routing si vous souhaitez uniquement la configuration en attribut PHP.

KeepAlive Messenger Transports 

Certains transports utilisés dans le composant Messenger ont un délai d’expiration qui détermine combien de temps un message est considéré comme « en cours de traitement ». Si un message n’est pas traité dans ce délai, il est renvoyé dans la file des messages en attente, ce qui peut potentiellement entraîner un traitement multiple du message.

Grâce à la contribution de Fabien Potencier et HypeMc, ce problème a été abordé de deux manières : 

Une nouvelle interface a été ajoutée, définissant la méthode suivante : 

interface KeepaliveReceiverInterface extends ReceiverInterface
{
    /**
     *  Informe le transport que le message est encore en cours de traitement, évitant ainsi un timeout coté transport
     *
     * @param int|null $seconds la durée minimum pour laquelle un message doit rester en keepalive
     *
     * @throws TransportException si il ya un problème avec le transport
     */
    public function keepalive(Envelope $envelope, ?int $seconds = null): void;
}

Un évènement ConsoleAlarmEvent a été ajouté pour notifier le signal \SIGALRM et permettre aux commandes console de programmer des alarmes. La classe principale Application, étendue par toutes les applications console Symfony, inclut désormais la méthode suivante : 

// Modifie l'intervalle pour programmer un signal SIGALRM toutes les n secondes. 
public function setAlarmInterval(?int $seconds): void 

Grâce à cette nouvelle fonctionnalité, les commandes messenger:consume et messenger:failed:retry ont été mises à jour pour inclure une nouvelle option  --keepalive. Cette amélioration permet aux commandes Messenger de suivre la logique suivante : 

private const DEFAULT_KEEPALIVE_INTERVAL = 5; 
// ... 
 
// programmé pour envoyer un SIGALRM toutes les n secondes (ici 5)
if ($input->hasParameterOption('--keepalive')) { 
   $this->getApplication()->setAlarmInterval( 
       (int) ($input->getOption('keepalive') ?? self::DEFAULT_KEEPALIVE_INTERVAL) 
   ); 
} 
 
// ... 
 
// Déclenché par l'évenement SIGALRM en appelant la fonctionnalité KeepAlive du transport
public function handleSignal(int $signal, int|false $previousExitCode = 0): int|false 
{ 
   // ... 
 
   if (\SIGALRM === $signal) { 
       $this->worker->keepalive(); 
       // ... 
   } 
} 

Désormais, vous pouvez lancer vos commandes de cette façon pour activer l’option keepalive (en supposant que le transport  « async » est toujours configuré dans votre messenger.yaml)

php bin/console messenger:consume async --keepalive
php bin/console messenger:failed:retry --keepalive

Symfony 7.2 marque une nouvelle étape dans l’évolution du composant Messenger.

L’introduction de la fonction KeepAlive améliore la fiabilité des traitements en évitant les pertes de messages et on peut se passer de la configuration YAML concernant le routing du transport grâce à PHP 8 et aux attributs !
Ces avancées vont nous permettre de gagner en productivité et de construire des applications plus robustes.

Sources

https://symfony.com/blog/new-in-symfony-7-2-keepalive-messenger-transports

https://symfony.com/blog/new-in-symfony-7-2-asmessage-attribute

https://symfony.com/doc/current/messenger.html