Temps de lecture estimé : 5 minutes.

RxJS est une librairie permettant de manipuler et gérer des traitements asynchrones. Elle est devenue incontournable lorsque l'on commence à jouer avec des frameworks Web tel qu'Angular qui l'utilisent massivement. Lorsque je joue avec une technologie, j'aime bien m'intéresser à cette problématique qu'est l'asynchronisme : je l'avais déjà fait pour C++ et je vous propose dans cet article de retrouver l'équivalent de nos petites habitudes avec la TPL .NET. 

Attention, l'idée de cet article n'est pas de faire une présentation de RxJS mais plutôt de vous donner quelques chemins de réflexion pour appréhender la philosophie du framework lorsque l'on a l'habitude de la TPL.

Voici ce que nous allons voir :

• Enchaîner 2 traitements asynchrones / mergeMap,
• Réaliser plusieurs traitements en parallèle,
• Gérer les erreurs / catch et retry,
• Faire durer un traitement à minima une certaine durée / combineLatest.

Enchaîner 2 traitements asynchrones / mergeMap

L'idée ici est de faire un premier traitement et d'utiliser le résultat de celui-ci pour déclencher un second traitement asynchrone. On produira alors une méthode fournissant le résultat de ce second traitement. La solution pour résoudre notre problème consiste à utiliser l'opérateur concatMap : il prend en paramètre une fonction retournant une autre Observable.

Le paramètre de cette fonction sera le résultat produit par le premier traitement asynchrone. Il sera nécessaire d'importer les opérateurs RxJs en amont de votre code. Attention, le traitement d'une seconde valeur produite éventuellement par le premier traitement ne sera déclenché qu'après le traitement de la première par toutes le valeurs émises par second traitement. 

import concatMap from 'rxjs/operators';

// une fonction qui retourne un nb aléatoire
const premierTraitement =
  () => Rx.Observable.of(Math.random());

// une fonction avec paramètre qui retourne un nb aléatoire
const secondTraitement =
  (param: number) => Rx.Observable.of(param + Math.random());

const traitementComplet = () =>
  // premier traitement déclenché
  premierTraitement()
  // on appelle l'opérateur mergemap
  .concatMap(
    // utilisation du résultat premierTraitement
    (resultat1) => secondTraitement(resultat1)
  );

Autre fait intéressant à noter, si ma deuxième opération n'est pas asynchrone, je dois alors utiliser l'opérateur map au lieu de concatMap :

import mergeMap from 'rxjs/operators';

// une fonction qui retourne un nb aléatoire
const premierTraitement =
  () => Rx.Observable.of(Math.random());

// une fonction avec paramètre qui retourne un nb aléatoire
const secondTraitement =
  (param: number) => Rx.Observable.of(param + Math.random());

const traitementComplet = () =>
  // premier traitement déclenché
  premierTraitement()
  // on appelle l'opérateur map
  .map(
    // utilisation du résultat premierTraitement
    (resultat1) => 10 + resultat1
  );

Gérer les erreurs / catch et retry

Avec RxJS, cela sera un tout petit peu plus complexe qu'un try/catch car cela est ... plus puissant. L'opérateur à utiliser sera catch et il prend en paramètre une fonction prenant elle-même comme entrées l'erreur produite par l'Observable originale et l'Observable originale elle-même. Cette fonction peut alors retourner plusieurs choses différentes (la quintessence de Javascript se retrouve dans cette dernière phrase) : soit une Observable, soit un tableau de valeurs directement produites soit lever une exception. Bien sûr, si aucune erreur ne se produit, le résultat original est produit par l'observable. 

// un observable qui leve une exception
const premierTraitement =
  () => Rx.Observable.of(Math.random())
    .map(val => { throw 'exception'; });


const traitementComplet = () =>
  // premier traitement déclenché
  premierTraitement()
    // catch de l'erreur
    .catch(
    // retour directement de la valeur
    (err, originalObservable) => [-1]
    );

Là où la RxJS devient intéressante c'est qu'il est possible de rejouer l'Observable originale simplement en la retournant. On a ainsi un mécanisme de retry assez facilement. Pour éviter les boucles infinies si l'erreur se produit tout le temps, je vous conseille cependant d'utiliser directement l'opérateur retry (essaie un certain nombre de fois) ou retryWhen (permet de fournir un prédicat conditionnant le ré-essai).

let throwException = true;
// un observable qui lève une exception une fois
//  (peut-être est-il Belge ?)
const premierTraitement =
  () => Rx.Observable.of(Math.random())
    .map(val => {
      if (throwException) {
        throwException = false;
        throw 'exception';
      }
      return val;
    });

const traitementComplet = () =>
  // premier traitement déclenché
  premierTraitement()
    // catch de l'erreur
    .catch(
    // retour directement de la valeur
    (err, original) => original
    );

Faire durer un traitement à minima une certaine durée

L'idée ici est de s'assurer qu'un traitement ne soit pas effectué instantanément. Cela permet par exemple de ne pas avoir des loaders qui "flashent" à l'écran quelques millisecondes pour disparaître. 

Nous allons utiliser l'opérateur combineLatest qui permet d'agréger le résultat de 2 Observables. Nous allons donc créer un Observable correspondant à l'attente minimum et le combiner avec notre Observable réel en retournant directement le résultat de celui-ci. L'intérêt de cette méthode réside dans le fait que l'abonnement à votre Observable sera bien fait immédiatement et le déclenchera donc directement lors de l'appel. Pratique par exemple lorsque vous utilisez l'HttpClient Angular qui ne fait concrètement l'appel que lors du Subscribe. Le code à écrire sera donc :

const premierTraitement = () => Rx.Observable.of(Math.random());

const traitementComplet = () => {
  // attente mininum de 2 secondes
  const delay = Rx.Observable.timer(2000);

  return premierTraitement()
    // combine l'attente et le traitement
    .combineLatest(delay,
          // on retourne le résultat de premierTraitement
          (a, b) => a);
};

PS: En .Net, j'aurais utilisé cette technique d'un précédent article.

Réaliser plusieurs traitements en parallèle / forkJoin

Avec RxJs, je vais utiliser l'opérateur forkJoin qui se comporte de manière très similaire à un Task.WhenAll. En s'abonnant à cet Observable, on récupérera un le tableau de résultats correspondant : 

const traitementComplet = () => {
  let tasks = [];
  for (let i = 0; i < 20; i++) {
    tasks.push(premierTraitement());
  }

  return Rx.Observable.forkJoin(tasks);
};

Happy coding.