Retour au Blog
Angular Development24 décembre 20259 min de lecture

Les opérateurs RxJS que vous utilisez vraiment en production avec Angular

DR

Doug Ribeiro

9 min de lecture

Je développe en Angular professionnellement depuis 2017. J'ai vu des équipes se noyer dans des chaînes d'opérateurs que personne ne pouvait debugger, et j'ai vu des équipes sous-utiliser RxJS et se retrouver avec des conditions de concurrence qui prenaient des semaines à traquer. L'écart entre les deux extrêmes est plus petit qu'on ne le pense. Il faut environ dix opérateurs, compris en profondeur.

switchMap : annuler et remplacer

switchMap est l'opérateur que j'utilise en premier quand une saisie utilisateur déclenche un appel HTTP. Quand une nouvelle valeur arrive de la source, switchMap se désabonne de l'observable interne précédent et en démarre un nouveau. Pour la recherche au fil de la frappe, cela signifie que la requête en cours de la frappe précédente est annulée et remplacée.

J'ai beaucoup utilisé cela sur la plateforme de cartes de crédit Wonderlabs. Les filtres de cartes et de transactions rechargeaient les données à chaque changement. Sans switchMap, les réponses arrivaient dans le désordre et écrasaient les résultats plus récents avec des résultats plus anciens. C'est un bug subtil, difficile à reproduire en production.

typescript 
@Component({ selector: 'app-product-search', standalone: true, template: '' })
export class ProductSearchComponent {
  private productService = inject(ProductService);
  searchControl = new FormControl('');
  results: Product[] = [];

  // In the constructor, so bare takeUntilDestroyed() has an injection context
  constructor() {
    this.searchControl.valueChanges.pipe(
      debounceTime(350),
      distinctUntilChanged(),
      switchMap(query => this.productService.search(query ?? '')),
      takeUntilDestroyed()
    ).subscribe(results => this.results = results);
  }
}

exhaustMap : ignorer tant que ce n'est pas terminé

exhaustMap est le sous-estimé. Tant qu'un observable interne est actif, il ignore toutes les nouvelles émissions de la source. Le cas d'usage canonique est un bouton de soumission de formulaire. Mettez exhaustMap derrière le bouton de soumission d'un formulaire de demande de prêt : si l'utilisateur double-clique ou clique à nouveau pendant l'appel API, le second clic est silencieusement ignoré. La requête n'est pas dupliquée.

typescript 
@Component({ selector: 'app-application-form', standalone: true, template: '' })
export class ApplicationFormComponent {
  private lendingApi = inject(LendingApiService);
  private router = inject(Router);
  submitClicks$ = new Subject<void>();

  constructor() {
    this.submitClicks$.pipe(
      exhaustMap(() => this.lendingApi.submitApplication(this.form.value).pipe(
        catchError(err => {
          this.handleError(err);
          return EMPTY;
        })
      )),
      takeUntilDestroyed()
    ).subscribe(confirmation => {
      this.router.navigate(['/confirmation', confirmation.id]);
    });
  }
}

Notez le catchError à l'intérieur de la fonction de projection de exhaustMap. Si l'erreur se propage au pipe externe, exhaustMap se termine et les clics futurs cessent de fonctionner. Il faut toujours capturer l'erreur à l'intérieur de l'observable interne.

mergeMap : tout exécuter en parallèle

Là où switchMap annule et exhaustMap ignore, mergeMap s'abonne à chaque observable interne de manière concurrente. Chaque émission de la source produit un abonnement actif. Utilisez-le quand l'ordre n'a pas d'importance et que vous voulez traiter toutes les émissions.

typescript 
// Upload multiple documents in parallel for a loan application
const uploads$ = from(selectedDocuments).pipe(
  mergeMap(doc => this.documentService.upload(doc).pipe(
    catchError(() => of({ id: doc.id, status: 'failed' }))
  ))
);

uploads$.subscribe(result => this.updateUploadStatus(result));

concatMap : mettre en file et préserver l'ordre

concatMap, c'est mergeMap avec une file d'attente. Il s'abonne à un observable interne à la fois et attend qu'il se termine avant de démarrer le suivant. Utilisez-le quand l'ordre fait partie du contrat : des uploads séquentiels que le serveur recolle ensuite, des événements d'audit qui doivent arriver dans l'ordre où l'utilisateur les a produits, un assistant qui enregistre chaque étape avant que la suivante ne commence. Le coût, c'est la contre-pression. Si la source émet plus vite que l'observable interne ne se termine, la file grossit sans limite et rien ne vous le signale. Si vous ne pouvez pas démontrer que la source est plus lente que le travail, utilisez plutôt mergeMap avec une limite de concurrence.

typescript 
@Component({ selector: 'app-chunked-upload', standalone: true, template: '' })
export class ChunkedUploadComponent {
  private uploadService = inject(UploadService);
  private destroyRef = inject(DestroyRef);

  send(chunks: Chunk[]): void {
    // Each chunk must land before the next one starts
    from(chunks).pipe(
      concatMap(chunk => this.uploadService.sendChunk(chunk)),
      takeUntilDestroyed(this.destroyRef)
    ).subscribe(ack => this.markChunkStored(ack.index));
  }

  sendUnordered(chunks: Chunk[]): void {
    // mergeMap with a concurrency cap when order does not matter but load does
    from(chunks).pipe(
      mergeMap(chunk => this.uploadService.sendChunk(chunk), 4),
      takeUntilDestroyed(this.destroyRef)
    ).subscribe();
  }
}
switchMap / mergeMapswitchMap : annule l'observable…switchMap : pour la recherche, …La dernière émission gagne. Les…mergeMap : exécute tous les obs…mergeMap : pour des requêtes pa…Risque : réponses dans le désor…exhaustMap / concatMapexhaustMap : ignore les nouvell…exhaustMap : pour la soumission…Empêche la double soumission. L…concatMap : met en file, exécut…concatMap : pour les uploads sé…Risque : la file peut croître s…VS

combineLatest : réagir à plusieurs flux simultanément

combineLatest émet chaque fois qu'une de ses sources émet, en combinant la dernière valeur de chacune. Je l'ai utilisé sur les tableaux de bord d'investissement en temps réel chez GFT. Un tableau de bord filtré par plage de dates, région et type de produit doit recharger les données à chaque changement de filtre.

typescript 
@Component({ selector: 'app-report-dashboard', standalone: true, template: '' })
export class ReportDashboardComponent {
  private reportService = inject(ReportService);
  reportData: ReportRow[] = [];

  constructor() {
    combineLatest([
      this.dateRangeControl.valueChanges.pipe(startWith(this.dateRangeControl.value)),
      this.regionControl.valueChanges.pipe(startWith(this.regionControl.value)),
      this.productControl.valueChanges.pipe(startWith(this.productControl.value))
    ]).pipe(
      debounceTime(200),
      switchMap(([dateRange, region, product]) =>
        this.reportService.fetch({ dateRange, region, product })
      ),
      takeUntilDestroyed()
    ).subscribe(data => this.reportData = data);
  }
}

Le startWith sur chaque contrôle est important. combineLatest n'émettra pas tant que toutes les sources n'ont pas émis au moins une fois. Sans startWith, le tableau de bord attend que l'utilisateur interagisse avec les trois filtres avant de charger quoi que ce soit.

forkJoin : attendre que tout soit terminé

forkJoin est le Promise.all des observables. Il attend que toutes les sources se complètent et émet leurs valeurs finales ensemble. Je l'utilise dans les resolvers de routes et la logique d'initialisation.

typescript 
@Component({ selector: 'app-lending-page', standalone: true, template: '' })
export class LendingPageComponent implements OnInit {
  private destroyRef = inject(DestroyRef);

  ngOnInit(): void {
    // Page init: load user, permissions, and feature config before rendering.
    // Outside the constructor, takeUntilDestroyed needs the DestroyRef passed in.
    forkJoin({
      user: this.userService.getCurrentUser(),
      permissions: this.authService.getPermissions(),
      config: this.featureService.getLendingConfig()
    }).pipe(
      takeUntilDestroyed(this.destroyRef)
    ).subscribe(({ user, permissions, config }) => {
      this.initializePage(user, permissions, config);
    });
  }
}

Un piège : si une source dans forkJoin rencontre une erreur, l'ensemble échoue. Encapsulez les observables individuels avec catchError si vous avez besoin de résultats partiels en cas d'échec.

debounceTime et distinctUntilChanged : toujours ensemble

Ces deux-là ne sont presque jamais utilisés séparément. debounceTime attend une période de calme. distinctUntilChanged ignore l'émission si la valeur n'a pas changé. Ensemble, ils empêchent le pattern d'abus d'API le plus courant : marteler un endpoint de recherche à chaque frappe avec la même valeur répétée.

typescript 
@Component({ selector: 'app-filter-panel', standalone: true, template: '' })
export class FilterPanelComponent {
  private api = inject(SearchApiService);
  filterControl = new FormControl('');
  results: SearchResult[] = [];

  constructor() {
    this.filterControl.valueChanges.pipe(
      debounceTime(400),
      distinctUntilChanged(),
      switchMap(term => this.api.filter(term ?? '')),
      takeUntilDestroyed()
    ).subscribe(results => this.results = results);
  }
}

Gestion des abonnements : takeUntilDestroyed

L'ancien pattern était un Subject, takeUntil et ngOnDestroy. Angular 16 a introduit takeUntilDestroyed depuis @angular/core/rxjs-interop, qui se connecte automatiquement au cycle de vie de destruction du composant. Sur n'importe quel codebase en Angular 16 ou plus, c'est le pattern que j'utilise.

typescript 
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { inject, DestroyRef } from '@angular/core';

@Component({ standalone: true, template: '' })
export class ExampleComponent {
  // Call outside of a constructor needs explicit DestroyRef
  private destroyRef = inject(DestroyRef);

  ngOnInit() {
    this.someObservable$.pipe(
      takeUntilDestroyed(this.destroyRef)
    ).subscribe();
  }

  // Or in constructor, no argument needed:
  constructor() {
    this.otherObservable$.pipe(
      takeUntilDestroyed() // Infers context automatically
    ).subscribe();
  }
}

L'opérateur que les gens oublient : le placement de catchError

Ce n'est pas un conseil sur un opérateur mais une erreur d'opérateur. Quand vous placez catchError au niveau du pipe externe et qu'une erreur survient, le flux entier se termine. Si ce flux alimente un élément de l'interface, il cesse de répondre pour toujours. Placez toujours catchError à l'intérieur de l'observable interne (dans switchMap, exhaustMap, etc.) et retournez EMPTY ou une valeur de repli. Le flux externe reste actif.

typescript 
// Wrong: error kills the entire search stream
this.searchControl.valueChanges.pipe(
  switchMap(q => this.api.search(q)),
  catchError(err => { // Stream terminates here. No more searches work.
    this.showError(err);
    return of([]);
  })
);

// Correct: error stays inside the inner observable
this.searchControl.valueChanges.pipe(
  switchMap(q => this.api.search(q).pipe(
    catchError(err => {
      this.showError(err);
      return of([]); // inner observable completes cleanly
    })
  ))
);

L'objectif n'est pas de connaître chaque opérateur RxJS. C'est d'en connaître un petit ensemble suffisamment en profondeur pour choisir le bon sans réfléchir.

switchMap pour les requêtes annulables. exhaustMap pour les soumissions de formulaires. mergeMap pour les opérations parallèles indépendantes. concatMap quand l'ordre fait partie du contrat. forkJoin pour le chargement de données au niveau de la page. combineLatest pour les tableaux de bord multi-filtres. debounceTime plus distinctUntilChanged sur chaque flux de saisie utilisateur. takeUntilDestroyed sur chaque abonnement. catchError à l'intérieur de l'observable interne. C'est la liste complète.

Références

AngularRxJSTypeScriptFrontend

Articles Connexes

Angular Development7 min de lecture

Debounce vs Throttle : la différence qui sauvera votre application Angular

Utiliser throttle sur un champ de recherche et debounce sur un gestionnaire de scroll sont tous deux des bogues. Voici comment ne plus jamais les confondre.

Angular Development9 min de lecture

Angular Signals vs NgRx : lequel devriez-vous utiliser ?

J'ai fait tourner NgRx sur une plateforme de cartes de crédit avec plus de 50 000 utilisateurs. La répartition des responsabilités est claire dès qu'on arrête de traiter cela comme une question binaire.

Angular Development10 min de lecture

Optimisation de performance Angular : ce qui fait vraiment la différence

J'ai audité beaucoup d'applications Angular et les mêmes lacunes apparaissent à chaque fois. Voici ce que je corrige en premier, pourquoi, et dans quel ordre.