Reactor Core

Introduction

Le principal cas d’usage de l’utilisation de Reactor Core est la création d’une application web, intranet ou extranet, avec Spring WebFlux, le pendant de Spring MVC en mode réactif.

Cependant, il peut être intéressant dans certains cas d’utiliser directement Reactor Core pour :

Nous allons aborder dans ce chapitre des éléments plutôt théoriques, car la librairie Spring Reactor Core est rarement utilisée seule pour plusieurs raisons :

Reactor Core

La programmation réactive est un concept spécial. Elle implique la propagation d’événements ou de signaux, comme des boules de flipper interagissant les unes avec les autres. En utilisant un style de programmation fonctionnelle, nous définissons les parcours possibles des boules, puis les lançons en parallèle pour qu’elles se déplacent simultanément. Chaque boule change de comportement lorsqu’elle interagit ou entre en collision avec d’autres éléments, ou lorsqu’elle atteint un point spécifique dans le temps. Le rôle du moteur est de gérer les interactions en appelant le code approprié pour chaque événement. Il les met en file d’attente dans une boucle d’événement et les exécute séquentiellement, ce qui donne l’impression d’une exécution parallèle depuis l’extérieur.

Les briques de base des Reactive Streams sont :

Méthode subscribe()

La méthode subscribe() est l’une des méthodes fondamentales en programmation réactive avec Reactor. Elle permet de déclencher l’exécution du flux réactif en s’abonnant à ce flux. Lorsqu’un flux est créé, il est inactif tant qu’aucune souscription n’est effectuée. La souscription déclenche l’émission des éléments du flux et l’exécution des traitements définis dans les différentes étapes du flux.

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);  
flux.subscribe(); 

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);  
flux.subscribe(item -> System.out.println("Élément : " + item)); 

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);  
flux.subscribe(  
    item -> System.out.println("Élément : " + item),  
    error -> System.err.println("Erreur : " + error.getMessage())  
); 

Utilisation du BaseSubscriber

En programmation réactive, nous utilisons généralement la programmation fonctionnelle avec les lambdas pour définir nos traitements. Cependant, dans certaines situations, il peut être utile de créer un Subscriber personnalisé pour gérer les événements d’un flux réactif de manière spécifique. C’est là que la classe utilitaire BaseSubscriber entre en jeu.

BaseSubscriber est une classe utilitaire fournie par Reactor qui simplifie la création de souscripteurs personnalisés pour les flux réactifs. Elle fournit une implémentation de base ainsi que des méthodes de gestion des événements-clés, ce qui facilite la personnalisation du comportement du souscripteur en fonction des besoins spécifiques de l’application. Pour créer un Subscriber personnalisé, il suffit d’étendre la classe BaseSubscriber et de substituer les méthodes appropriées en fonction de nos besoins. Par exemple, en implémentant la méthode hookOnNext(), nous pouvons spécifier l’action à effectuer lorsqu’un nouvel élément est émis par le flux. De même, en implémentant la méthode hookOnError(), nous pouvons gérer les erreurs qui surviennent pendant le traitement du flux.

Il est important de noter que les instances de BaseSubscriber (ou de ses sous-classes) sont à usage unique. Cela signifie qu’un BaseSubscriber annulera automatiquement son abonnement au premier Publisher s’il est abonné à un second Publisher. Cette gestion automatique garantit que le souscripteur est utilisé de manière appropriée dans le contexte réactif.

Le BaseSubscriber de Reactor propose plusieurs...

Générer une séquence manuellement

Nous pouvons générer les séquences de façon synchrone ou asynchrone.

Flux<Integer> infiniteSequence = Flux.generate(  
    () -> 0, // État initial  
    (state, sink) -> {  
        sink.next(state); // Émet l'élément dans le flux  
        return state + 1; // Met à jour l'état  
    }  
); 

Méthode handle()

La méthode handle() de la classe Flux est utilisée pour gérer de manière personnalisée les éléments émis par le flux. Elle permet de transformer les éléments du flux ou d’effectuer des traitements spécifiques sur chaque élément. La signature de la méthode handle() est la suivante :

Flux<T> handle(BiConsumer<T, SynchronousSink<R>> handler) 

Le paramètre handler est un BiConsumer qui prend deux arguments : l’élément émis par le flux (T) et un SynchronousSink<R> utilisé pour émettre les éléments résultants du traitement (R). Le SynchronousSink est un objet qui permet de générer de manière synchrone des éléments dans un flux. Il est utilisé pour émettre les éléments résultants du traitement à partir du handler. Dans le handler, vous pouvez effectuer différentes actions sur l’élément reçu, telles que le transformer en un nouvel élément (R) à émettre dans le flux, l’ignorer, le filtrer ou effectuer toute autre logique de traitement nécessaire.

Voici un exemple :

Flux<Integer> originalFlux = Flux.range(1, 10);  
  
Flux<String> processedFlux = originalFlux.handle((number...

Scheduler

Spring Reactor offre aux développeurs plusieurs possibilités pour gérer l’ordonnancement des tâches grâce à l’utilisation d’ordonnanceurs (schedulers), aussi appelés planificateurs. Ceux-ci permettent de contrôler l’exécution des tâches de manière asynchrone et efficace. Ils s’utilisent avec les méthodes publishOn() et subscribeOn() pour contrôler le contexte d’exécution dans lequel les opérations du flux sont effectuées. Ces méthodes permettent de spécifier les schedulers à utiliser pour différentes parties du flux, ce qui peut avoir un impact significatif sur la concurrence et les performances de l’application.

La méthode publishOn() permet de spécifier le scheduler sur lequel les opérations suivantes seront exécutées. Ainsi, toutes les opérations qui viennent après l’appel à publishOn() seront effectuées dans le contexte du scheduler spécifié. Cela peut être utile pour déléguer le traitement asynchrone à un autre thread, particulièrement pour les opérations coûteuses en temps ou bloquantes. 

Par exemple :

    Flux.range(1, 10)  
      .publishOn(Schedulers.boundedElastic()) // Utilise le 
scheduler boundedElastic() pour le traitement asynchrone   
      .map(value -> {  
        // Code de traitement asynchrone  ...

Traiter correctement les cas d’erreurs

Avec Spring Reactor Core, vous pouvez gérer les erreurs de différentes manières. Voici quelques approches courantes pour gérer les erreurs dans un flux.

Flux<Integer> flux = Flux.just(1, 2, 3)  
    .map(i -> {  
        if (i == 2) {  
            throw new RuntimeException("Erreur !");  
        }  
        return i;  
    })  
    .onErrorResume(e -> Flux.just(10, 20, 30));  
  
flux.subscribe(  
    data -> System.out.println("Donnée : " + data),  
    error -> System.err.println("Erreur : " + error),  
    () -> System.out.println("Terminé")  
); 

Capture et renvoi d’exceptions

public String convert(int i) throws IOException {  
    if (i > 3) {  
        throw new IOException("boom " + i);  
    }  
    return "OK " + i;  
}  
  
Flux<String> converted = Flux  
    .range(1, 10)  
    .map(i -> {  
        try { return convert(i); }  
        catch (IOException e) { throw Exceptions.propagate(e); } 
    });  ...

Sinks

En Reactor, un sink offre la possibilité de déclencher manuellement des signaux de manière indépendante. Il crée ainsi une structure similaire à celle d’un éditeur de publication (Publisher) capable de gérer plusieurs abonnés (Subscriber), à l’exception des variantes unicast (communication de un vers un).

Les sinks de base exposés par Reactor Core garantissent que l’utilisation multithread est détectée et ne peut pas entraîner de violations de spécifications ou de comportements indéfinis du point de vue des abonnés en aval. Lorsque vous utilisez l’API tryEmit*, les appels parallèles échouent rapidement. Lorsque vous utilisez l’API emit*, l’EmissionFailureHandler fourni peut permettre de réessayer en cas de contentieux (par exemple, boucle occupée), sinon le sink se terminera avec une erreur. Cela représente une amélioration par rapport à Processor.onNext(), qui doit être synchronisé de manière externe ou peut entraîner un comportement indéfini du point de vue des abonnés en aval.

La classe Sinks est généralement privilégiée par rapport aux objets Processor, car elle offre une approche plus sûre et plus simple pour la gestion de signaux. Les constructeurs de sinks fournissent une API guidée pour les principaux types de producteurs pris en charge. Certains comportements trouvés dans Flux, tels que onBackpressureBuffer(), sont également présents dans dans la classe Sinks.

On trouve plusieurs catégories de sinks :...

Transformations personnalisées sur le flux

Il existe un certain nombre d’opérateurs qui sont utilisés pour effectuer des transformations personnalisées sur un flux :

Voici un exemple simple utilisant l’opérateur transform() pour transformer un flux de chaînes en un flux de leurs longueurs :

public class TransformExample {  
  public static void main(String[] args) {  
    // Créer un flux de chaînes  
    Flux<String> stringFlux = Flux.just("rouge", "vert", "orange"); 
  
    // Transformer le flux en un flux de longueurs de chaînes  
    Flux<Integer> lengthFlux = 
stringFlux.transform(stringFluxTransformer()); 
  
    // Souscrire au flux de longueurs  
    lengthFlux.subscribe(length -> 
System.out.println("Longueur : " + length));  
  }  
  
  private static Function<? super Flux<String>,? 
extends Publisher<Integer>> stringFluxTransformer() {  
    return flux -> flux.map(s -> s.length());  
  }  ...

Global hooks

Les global hooks (crochets globaux) sont des points d’extension qui permettent aux développeurs de modifier le comportement global de certains aspects-clés du framework. Autrement dit, ils centralisent la personnalisation de fonctionnalités qui s’appliquent à l’ensemble du flux réactif de l’application.

  public void onNextDroppedFailReplaces() {  
    AtomicReference<Object> dropHook = new AtomicReference<>(); 
    Publisher<Integer> p = s -> {  
      s.onSubscribe(Operators.emptySubscription());  
      s.onNext(1);  
      s.onNext(2);  
      s.onNext(3);  
    };  
    List<Integer> seen = new ArrayList<>();  ...

API Context

L’API Context est utilisée pour propager des informations de manière réactive à travers les Flux et les Mono. Elle permet de partager des données entre les différentes étapes du traitement réactif sans avoir à utiliser des variables globales ou des références partagées. La classe Context représente un contexte associé à un flux ou un mono. Elle est utilisée pour stocker des données sous forme de paires clé-valeur. Chaque flux ou mono peut avoir son propre contexte, et lorsqu’un flux ou un mono émet une valeur, il peut également propager son contexte aux opérateurs suivants.

Tests

L’approche TDD (Test Driven Development, soit développement piloté par les tests) exige des tests unitaires et des tests d’intégration Spring.

Nous commençons par ajouter la dépendance reactor-test :

<dependency>  
    <groupId>io.projectreactor</groupId>  
    <artifactId>reactor-test</artifactId>  
    <scope>test</scope>  
    (1)  
</dependency> 

Cette dépendance offre des fonctionnalités supplémentaires pour faciliter le test des applications basées sur Spring Reactor. Les principaux apports de cette dépendance sont les suivants :

La dépendance reactor-test fournit des outils puissants pour tester et valider le comportement des Flux et des Mono dans le contexte de tests unitaires...

Réacteur de débogage

Le mode de débogage de Reactor (Reactor Debug) est une fonctionnalité de Spring Reactor Core qui permet de faciliter le débogage et la compréhension du flux des événements lors du développement d’applications réactives. Il offre des outils pour visualiser les interactions entre les différents opérateurs et souscripteurs, ainsi que pour surveiller l’état du flux à différents moments de son exécution.

Pour activer le mode de débogage, vous pouvez utiliser les méthodes doOnSubscribe(), doOnNext(), doOnError(), doOnComplete() et doOnTerminate() avec l’opérateur log() qui affiche des informations de débogage à chaque étape du flux.

Voici un exemple d’utilisation des opérations doOnSubscribe(), doOnNext(), doOnError(), doOnComplete(), doOnTerminate() et log() pour activer le débogage dans un projet Reactor.

Flux.just("A", "B", "C", "D")  
  .log() // Affiche les informations de log  
  // Déclenché à la souscription  
  .doOnSubscribe(subscription -> System.out.println("Flux  
subscribed!"))  
  // Déclenché à chaque nouvel élément  
  .doOnNext(element -> System.out.println("Next element: " + 
element))  
  // Déclenché en cas d'erreur  
  .doOnError(error -> System.out.println("An error has occurred: " 
+ error.getMessage()))  
  // Déclenché lorsque le flux se termine normalement  
  .doOnComplete(() -> System.out.println("Flux complete!"))  
  // Déclenché lorsque le flux est terminé pour une quelconque  
autre raison 
  .doOnTerminate(() -> System.out.println("Flux termination  
(either onComplete or onError)!"))  
  .subscribe(); 

Ce programme...

Conclusion

En conclusion, l’utilisation de Reactor Core offre une approche puissante et flexible pour la programmation réactive en Java. Grâce à sa mise en œuvre des spécifications Reactor et Reactive Streams, il permet de gérer efficacement des flux de données asynchrones, tout en fournissant un ensemble complet d’opérateurs pour la transformation, la combinaison et la gestion des flux.

Avec Reactor Core, les développeurs peuvent créer des pipelines de traitement de données réactives, où les traitements sont effectués de manière asynchrone et non bloquante, permettant ainsi une meilleure utilisation des ressources système et une extensibilité accrue. Les opérateurs fournis permettent de traiter des erreurs, de gérer la contre-pression et de contrôler l’ordonnancement des tâches, ce qui facilite le développement d’applications réactives robustes et performantes.

Les schedulers offrent également un contrôle précis sur l’exécution des tâches, permettant de spécifier sur quel thread ou scheduler les opérations doivent être effectuées, ce qui facilite la gestion de la concurrence et de la parallélisation.

En outre, Reactor Core fournit des mécanismes pour gérer les abonnements et la durée de vie des flux, notamment...