RxJS avancé : opérateurs d’ordre supérieur

Les signals ont repris l’état, mais les événements dans le temps restent le terrain de RxJS : frappes au clavier, clics, requêtes qui se chevauchent, retries, websockets. La partie la plus importante, et celle que l’on rate le plus souvent, ce sont les opérateurs d’ordre supérieur. Chaque valeur démarre un nouvel Observable, généralement une requête HTTP. La vraie question est : que faites-vous quand la valeur suivante arrive alors que la requête précédente est encore en cours ?

Il y a quatre réponses : annuler, exécuter en parallèle, mettre en file ou ignorer. Chacune a son opérateur. Si vous choisissez mal, vous obtenez un bug de race condition : d’anciens résultats qui écrasent les nouveaux, des commandes en double, des enregistrements appliqués dans le mauvais ordre.

Le problème : un Observable d’Observables

Un map qui renvoie une requête produit un Observable<Observable<T>> : un flux de requêtes auxquelles personne ne s’est abonné. Les opérateurs *Map s’abonnent eux-mêmes aux flux internes et les aplatissent en un seul flux de résultats. La seule différence entre eux est la politique de chevauchement.

// « Pour chaque valeur, démarrer un Observable » — la question est de savoir ce que vous faites quand la valeur suivante arrive
// alors que la précédente tourne encore.
clicks$.pipe(map((id) => http.get(`/api/items/${id}`)));       // ✗ Observable<Observable<Item>>
clicks$.pipe(switchMap((id) => http.get(`/api/items/${id}`))); // ✓ Observable<Item> — plus une politique

Les quatre opérateurs

// Typeahead : seule la dernière requête compte ; les anciennes sont ANNULÉES (HttpClient annule aussi le XHR/fetch)
readonly results = toSignal(
  this.query$.pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((q) => this.api.search(q).pipe(catchError(() => of([])))),
  ),
  { initialValue: [] },
);

Où placer catchError

Une erreur termine le flux dans lequel elle se produit. Si catchError se trouve après switchMap, la première requête en échec ferme tout le flux : la recherche ne répond plus, sans aucune erreur dans la console. Placez-le à l’intérieur de la projection interne. Ainsi, seule cette requête est remplacée par une valeur de repli, et le flux externe continue.

// ✗ catchError APRÈS switchMap : la première erreur termine le flux externe — la recherche « meurt »
query$.pipe(
  switchMap((q) => api.search(q)),
  catchError(() => of([])),
);

// ✓ catchError DANS la projection : seule cette requête échoue, le flux externe reste vivant
query$.pipe(
  switchMap((q) => api.search(q).pipe(catchError(() => of([])))),
);

Retry avec backoff exponentiel

retry({ count, delay }) se réabonne à la source après une erreur. La fonction delay reçoit l’erreur et le numéro de la tentative, et renvoie un Observable qui dit quand réessayer. Doubler la pause à chaque étape (300, 600, 1200 ms…) évite de marteler le serveur quand il a des problèmes. Relancez l’erreur pour les codes qui ne se corrigeront pas d’eux-mêmes (401, 404). Pour HTTP, le retry vit généralement dans un interceptor (voir HttpClient & interceptors). Pour une connexion qui doit survivre aux redémarrages du serveur — avec du jitter, pour que mille clients ne reviennent pas dans la même seconde —, voir WebSockets.

import { retry, timer } from 'rxjs';

http.get('/api/report').pipe(
  retry({
    count: 4,                                                // 4 retries au maximum
    delay: (error, retryCount) => {
      if (error.status === 401 || error.status === 404) throw error; // ne pas réessayer les erreurs « définitives »
      return timer(300 * 2 ** (retryCount - 1));              // 300, 600, 1200, 2400 ms
    },
    resetOnSuccess: true,
  }),
);

share et shareReplay

Un Observable « cold », comme http.get, s’exécute à chaque subscribe. Trois abonnés, c’est trois requêtes. share() fait partager une seule exécution aux abonnés simultanés. shareReplay conserve en plus la dernière valeur pour ceux qui arrivent plus tard : il fonctionne donc comme un cache. Le piège est un simple shareReplay(1) : il a refCount: false, donc il ne se désabonne jamais de la source. Sur un flux qui ne se termine jamais (un websocket, un interval), la source continue de tourner après le départ de tous les abonnés.

Un détail subtil avec refCount: true : le cache n’est conservé que si la source a eu le temps de se terminer. Si tous les abonnés partent avant complete (par exemple firstValueFrom, take(1), ou un composant détruit pendant la requête), shareReplay se désabonne de la source et repart de zéro avec l’abonné suivant.

// Sans partage : chaque subscribe = une exécution de plus (une requête de plus)
const user$ = http.get<User>('/api/me');
user$.subscribe(); user$.subscribe();                 // 2 requêtes

// share() : les abonnés SIMULTANÉS partagent une exécution ; après complete, il se réinitialise
const shared$ = http.get<User>('/api/me').pipe(share());

// shareReplay : conserve la dernière valeur pour les abonnés TARDIFS (cache)
const cached$ = http.get<User>('/api/me').pipe(shareReplay({ bufferSize: 1, refCount: true }));
// refCount: true → quand le dernier abonné part, il se désabonne aussi de la source.
// shareReplay(1) (refCount: false) sur une source qui ne se termine jamais = la source tourne indéfiniment.

Combiner des flux

combineLatest émet dès qu’une source change, une fois que chacune a émis au moins une fois. Il convient aux filtres combinés. withLatestFrom n’émet qu’à partir de la source principale et prend la valeur courante des autres (« au clic, enregistrer le formulaire actuel »). forkJoin attend que toutes se terminent et n’émet qu’une fois : c’est un Promise.all pour les requêtes.

// combineLatest : émet à chaque changement, une fois que TOUTES ont émis une fois — filtres combinés
combineLatest([category$, sort$]).pipe(switchMap(([c, s]) => api.list(c, s)));

// withLatestFrom : n’émet qu’à partir de la source principale, en prenant la dernière valeur des autres
save$.pipe(withLatestFrom(form$), concatMap(([, value]) => api.save(value)));

// forkJoin : attend que TOUTES se terminent, une seule émission — requêtes parallèles indépendantes
forkJoin({ user: api.me(), settings: api.settings() });

// zip : associe la n-ième valeur de chacune — rarement utile dans une UI

Aucun abonnement oublié

Chaque subscribe manuel dans un composant doit être fermé. takeUntilDestroyed() le fait automatiquement à la destruction du composant. Pour les valeurs affichées dans le template, toSignal() est encore plus simple : il s’abonne et se désabonne tout seul (voir Signals ↔ RxJS).

import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

export class Live {
  constructor() {
    interval(1000)
      .pipe(takeUntilDestroyed()) // dans un contexte d’injection ; sinon : takeUntilDestroyed(this.destroyRef)
      .subscribe((n) => this.tick.set(n));
  }
}
// Pour l’état affiché dans le template, toSignal(obs$) se désabonne tout seul (voir la page Signals ↔ RxJS).

Quand OUI, quand NON

Pièges fréquents

• mergeMap pour la recherche. Les réponses arrivent dans n’importe quel ordre, et une ancienne peut écraser la nouvelle. Utilisez switchMap.
• switchMap pour les enregistrements. Un enregistrement annulé à mi-chemin a peut-être déjà été traité par le serveur, ou non. Pour les écritures, utilisez concatMap ou exhaustMap. Plus de détails sur les écritures qui arrivent dans le désordre dans les mises à jour optimistes.
• Subscribe dans subscribe. Vous perdez l’annulation et la gestion des erreurs. Tout « subscribe, puis un autre subscribe dans le callback » est un opérateur *Map.
• catchError au niveau externe d’un flux qui doit rester vivant : il arrête tout à la première erreur.
• shareReplay(1) sur des sources infinies : fuites de mémoire et de connexions. Utilisez { bufferSize: 1, refCount: true }. Comment les trouver : voir la page sur les fuites de mémoire.
• Retry sur 4xx : vous réessayez indéfiniment une requête qui ne réussira jamais. Filtrez par statut.

Aide-mémoire

OPÉRATEUR     QUAND UNE NOUVELLE VALEUR ARRIVE PENDANT QUE L’ANCIENNE TOURNE  USAGE
switchMap     annuler l’ancienne                                              recherche, paramètres de route, « la dernière gagne »
mergeMap      les exécuter en parallèle (optionnel : concurrency)             opérations indépendantes, envois de fichiers
concatMap     la mettre en file, dans l’ordre                                 enregistrements ordonnés, files de commandes
exhaustMap    l’ignorer                                                       envoi de formulaire, connexion, rafraîchissement manuel

catchError    DANS la projection interne, pour que le flux externe reste vivant
retry         { count, delay: (err, n) => timer(base * 2 ** (n - 1)) } — ne pas réessayer les 4xx
share()       les abonnés simultanés partagent l’exécution
shareReplay   { bufferSize: 1, refCount: true } — cache + désabonnement correct
combineLatest filtres combinés · withLatestFrom « au clic, prendre la valeur courante » · forkJoin « toutes, une fois »
takeUntilDestroyed() / toSignal()   aucun abonnement oublié