RxJS: programación reactiva

Un observable describe una fuente de valores que llegan con el tiempo: clics, respuestas HTTP, ticks de un temporizador. Emite con next cero, uno o muchos valores y termina de una de dos formas: con complete, si acaba bien, o con error, si falla. Después de cualquiera de las dos, no emite nada más.

Es perezoso: definirlo no hace nada. El trabajo empieza al llamar a subscribe(), y cada suscripción vuelve a ejecutar el productor desde cero. Por eso, si te suscribes dos veces a una petición de HttpClient, salen dos peticiones.

subscribe() devuelve una suscripción: al llamar a unsubscribe() dejas de recibir valores y el productor libera lo que estuviera usando.

Un operador convierte un flujo en otro. Se encadenan con pipe(), como en una cadena de montaje: map transforma cada valor, filter deja pasar solo los que cumplen una condición y tap ejecuta algo por cada valor sin cambiarlo, muy útil para depurar.

Dos operadores de tiempo aparecen en casi todos los buscadores: debounceTime(300) espera a que pasen 300 ms sin valores nuevos y emite solo el último, y distinctUntilChanged() descarta un valor si es igual al anterior.

Cuando llega un valor nuevo, switchMap cancela el observable interior anterior (se desuscribe de él) y se cambia al nuevo. Solo te llegan los resultados de la petición más reciente.

Con HttpClient, desuscribirse cancela de verdad la petición en el navegador: no gastas red en respuestas que ya no te interesan y evitas que una respuesta antigua y lenta pise a una nueva.

mergeMap se suscribe a cada observable interior en cuanto llega su valor y deja que corran a la vez. Los resultados salen en el orden en que terminan, no en el orden en que empezaron.

Admite un segundo argumento para limitar la concurrencia: mergeMap(fn, 3) mantiene como mucho tres en marcha y deja el resto en espera.

concatMap procesa los observables interiores de uno en uno, en el orden en que llegaron sus valores. Mientras uno está en marcha, los siguientes esperan en una cola; no se cancela ni se pierde ninguno.

Es la opción segura para operaciones que deben aplicarse en orden, como varias escrituras sobre el mismo recurso.

exhaustMap hace lo contrario que switchMap: mientras el observable interior actual no termina, ignora los valores que lleguen. Cuando termina, vuelve a escuchar.

Es el antídoto contra el doble clic: aunque el usuario pulse el botón tres veces seguidas, solo sale una petición.

Un Subject es a la vez observable y emisor: llamas a next(valor) desde fuera y todos sus suscriptores lo reciben. Un Subject normal no recuerda nada, así que quien se suscribe tarde se pierde lo anterior. Sirve para eventos, como «se ha cerrado la sesión».

BehaviorSubject necesita un valor inicial, recuerda el último y se lo entrega al instante a cada suscriptor nuevo; además, puedes leerlo con .value. ReplaySubject(n) recuerda los últimos n valores y se los repite a quien llega.

Antes de los signals, un servicio con un BehaviorSubject privado y un asObservable() público era la forma típica de compartir estado en Angular. Hoy, para estado, un signal suele ser más sencillo.

Una suscripción a un flujo que no termina sigue viva aunque el componente se destruya: el código se sigue ejecutando, retiene memoria y puede intentar actualizar una pantalla que ya no existe. Hay que cancelarla, y lo mejor es no hacerlo a mano.

El pipe async se suscribe desde la plantilla y se desuscribe solo cuando se destruye el componente. Si te suscribes en la clase, takeUntilDestroyed(), de @angular/core/rxjs-interop, completa el flujo al destruirse el componente (fuera del constructor, pásale un DestroyRef). Y toSignal también gestiona la suscripción por ti.

combineLatest([a$, b$]) emite un array con el último valor de cada fuente cada vez que cualquiera de ellas emite, pero solo cuando todas han emitido al menos una vez. Es ideal para combinar filtros que cambian, como la categoría y el orden.

forkJoin([a$, b$]) espera a que todas terminen y emite una sola vez con el último valor de cada una, como Promise.all. Es perfecto para cargar en paralelo varias peticiones independientes antes de pintar una pantalla.

Cuando un observable emite un error, termina. catchError intercepta ese error y lo sustituye por otro observable: un valor por defecto con of([]), o un error más claro con throwError.

retry vuelve a suscribirse si hay un fallo, es decir, repite la petición. Con un objeto de configuración decides cuántas veces y cuánto esperar entre intentos: retry({ count: 3, delay: 1000 }).