Volver al Blog
Angular Development24 de diciembre de 20259 min de lectura

Operadores RxJS que realmente usas en aplicaciones Angular en producción

DR

Doug Ribeiro

9 min de lectura

Escribo Angular profesionalmente desde 2017. He visto equipos ahogarse en cadenas de operadores que nadie podía depurar, y he visto equipos subutilizar RxJS y terminar con condiciones de carrera que tomaron semanas rastrear. La distancia entre ambos extremos es menor de lo que la gente cree. Necesitas unos diez operadores, entendidos a fondo.

switchMap: Cancelar y reemplazar

switchMap es el operador al que recurro primero cuando la entrada del usuario dispara una llamada HTTP. Cuando llega un nuevo valor desde la fuente, switchMap cancela la suscripción al observable interno anterior y comienza uno nuevo. Para búsqueda mientras se escribe, esto significa que la solicitud en vuelo del keystroke anterior se cancela y se reemplaza.

Usé esto extensivamente en la plataforma de tarjetas de crédito de Wonderlabs. Los filtros de tarjetas y transacciones volvían a consultar con cada cambio. Sin switchMap, las respuestas llegaban fuera de orden y sobreescribían resultados más recientes con otros más antiguos. Ese es un bug sutil y difícil de reproducir en producción.

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: Ignorar hasta que termine

exhaustMap es el subestimado. Mientras un observable interno está activo, ignora todas las nuevas emisiones de la fuente. El caso de uso canónico es un botón de envío de formulario. Pon exhaustMap detrás del botón de envío de un formulario de solicitud de préstamo: si el usuario hace doble clic o vuelve a hacer clic durante la llamada a la API, el segundo clic se ignora silenciosamente. La solicitud no se duplica.

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]);
    });
  }
}

Observa el catchError dentro de la función de proyección de exhaustMap. Si el error se propaga al pipe externo, exhaustMap termina y los futuros clics dejan de funcionar. Siempre captura dentro del observable interno.

mergeMap: Ejecutar todo en paralelo

Donde switchMap cancela y exhaustMap ignora, mergeMap se suscribe a cada observable interno de forma concurrente. Cada emisión de la fuente produce una suscripción activa. Usa esto cuando el orden no importa y quieres que todas las emisiones se manejen.

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: Encolar y preservar el orden

concatMap es mergeMap con una cola. Se suscribe a un observable interno a la vez y espera a que termine antes de arrancar el siguiente. Úsalo cuando el orden es parte del contrato: subidas secuenciales que el servidor vuelve a unir, eventos de auditoría que deben llegar en el orden en que el usuario los produjo, un wizard que guarda cada paso antes de que empiece el siguiente. El costo es la contrapresión. Si la fuente emite más rápido de lo que el observable interno termina, la cola crece sin límite y nada te avisa. Si no puedes argumentar que la fuente es más lenta que el trabajo, usa mergeMap con un límite de concurrencia en su lugar.

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: cancela el observabl…switchMap: úsalo para búsqueda …Gana la última emisión. Las ant…mergeMap: ejecuta todos los obs…mergeMap: úsalo para peticiones…Riesgo: respuestas fuera de ord…exhaustMap / concatMapexhaustMap: ignora las nuevas m…exhaustMap: úsalo para submit d…Evita el doble submit. Gana la …concatMap: encola y ejecuta en …concatMap: úsalo para uploads s…Riesgo: la cola puede crecer si…VS

combineLatest: Reaccionar a múltiples flujos a la vez

combineLatest emite cada vez que cualquiera de sus fuentes emite, combinando el último valor de cada una. Lo usé en los dashboards de inversión en tiempo real de GFT. Un dashboard filtrado por rango de fechas, región y tipo de producto necesita volver a consultar cada vez que cualquier filtro cambia.

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);
  }
}

El startWith en cada control es importante. combineLatest no emitirá hasta que todas las fuentes hayan emitido al menos una vez. Sin startWith, el dashboard espera a que el usuario interactúe con los tres filtros antes de cargar algo.

forkJoin: Esperar a que todos completen

forkJoin es el Promise.all de los observables. Espera a que todas las fuentes completen y emite sus valores finales juntos. Lo uso en resolvers de rutas y lógica de inicialización.

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 detalle importante: si cualquier fuente en forkJoin lanza un error, todo falla. Envuelve los observables individuales con catchError si necesitas resultados parciales en caso de fallo.

debounceTime y distinctUntilChanged: Siempre juntos

Estos dos casi nunca se usan por separado. debounceTime espera un periodo de inactividad. distinctUntilChanged descarta la emisión si el valor no ha cambiado. Juntos previenen el patrón más común de abuso de API: bombardear un endpoint de búsqueda en cada keystroke con el mismo valor repetido.

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);
  }
}

Gestión de suscripciones: takeUntilDestroyed

El patrón antiguo era un Subject, takeUntil y ngOnDestroy. Angular 16 introdujo takeUntilDestroyed desde @angular/core/rxjs-interop, que se engancha automáticamente al ciclo de vida de destrucción del componente. En cualquier codebase que corra Angular 16 o posterior, este es el patrón que uso.

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();
  }
}

El operador que la gente olvida: la ubicación de catchError

Esto no es un consejo de operador sino un error de operador. Cuando pones catchError en el nivel del pipe externo y ocurre un error, todo el flujo se termina. Si ese flujo controla un elemento de UI, deja de responder para siempre. Siempre pon catchError dentro del observable interno (dentro de switchMap, exhaustMap, etc.) y retorna EMPTY o un valor de respaldo. El flujo externo permanece vivo.

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
    })
  ))
);

El objetivo no es conocer cada operador de RxJS. Es conocer un conjunto pequeño lo suficientemente a fondo como para elegir el correcto sin pensarlo.

switchMap para solicitudes cancelables. exhaustMap para envíos de formularios. mergeMap para operaciones paralelas independientes. concatMap cuando el orden es parte del contrato. forkJoin para carga de datos a nivel de página. combineLatest para dashboards con múltiples filtros. debounceTime más distinctUntilChanged en cada flujo de entrada del usuario. takeUntilDestroyed en cada suscripción. catchError dentro del observable interno. Esa es la lista completa.

Referencias

AngularRxJSTypeScriptFrontend

Publicaciones Relacionadas

Angular Development7 min de lectura

Debounce vs Throttle: La Diferencia Que Salvará Tu App Angular

Usar throttle en un input de búsqueda y debounce en un handler de scroll son ambos bugs. Así es como no volverás a confundirlos.

Angular Development9 min de lectura

Angular Signals vs NgRx: ¿Cuál deberías usar?

Corrí NgRx en una plataforma de tarjetas de crédito con más de 50 000 usuarios. La división de responsabilidades es clara una vez que dejas de tratar esto como una pregunta de uno u otro.

Angular Development10 min de lectura

Optimización de rendimiento en Angular: Lo que realmente marca la diferencia

He auditado muchas apps Angular y las mismas brechas aparecen cada vez. Esto es lo que arreglo primero, por qué, y en qué orden.