Módulo 14: RxJS fundamental

Objetivos del módulo

• Entender qué es un flujo de valores en el tiempo y qué problema resuelve RxJS frente a las promesas y las señales
• Crear un Observable, suscribirte con un Observer y cortar la ejecución con Subscription
• Leer y dibujar diagramas de canicas
• Generar flujos con of, from, interval, timer, fromEvent, defer y throwError
• Transformar y filtrar con pipe, map, filter, tap, scan, take, distinctUntilChanged y compañía
• Elegir con criterio entre switchMap, mergeMap, concatMap y exhaustMap
• Frenar eventos con debounceTime, throttleTime, auditTime y sampleTime
• Combinar flujos con combineLatest, forkJoin, withLatestFrom, zip, merge, concat y race
• Tratar errores con catchError, retry, timeout y finalize, y saber dónde colocar cada uno
• Usar Subject, BehaviorSubject, ReplaySubject y AsyncSubject
• Distinguir flujos fríos y calientes, y compartir una ejecución con share y shareReplay
• Detectar y evitar fugas de suscripciones
• Convertir entre observables y promesas con firstValueFrom y lastValueFrom
• Escribir operadores propios y probar flujos con canicas y TestScheduler

📘 Concepto: Este módulo trata de RxJS, una biblioteca independiente de Angular. El curso usa RxJS 7.8.2, la misma versión que instala ng new en Angular 22 y en la LTS, así que todo funciona igual en ambas. Solo hay una marca, en la sección 6, por un detalle de HttpClient. Los servicios de los ejemplos usan @Injectable({ providedIn: 'root' }) para que compilen sin cambios en las dos versiones. En Angular 22 puedes escribir @Service(), como viste en el módulo 08.

---

1. Por qué existe RxJS

Ya sabes manejar dos clases de datos que cambian. Una señal (módulo 05) guarda un valor actual: lo que importa es cuánto vale ahora. Una promesa representa un único resultado que llegará más tarde. Ninguna de las dos describe bien esta situación:

El usuario escribe en un buscador. Cada tecla debería lanzar una búsqueda, pero no si sigue escribiendo; si escribe otra letra, la búsqueda anterior ya no interesa y hay que cancelarla; si una falla, el buscador tiene que seguir funcionando.

Aquí lo importante no es el valor actual, sino la secuencia de sucesos en el tiempo y cómo se relacionan: cuándo llegan, cuáles se descartan y cuáles cancelan a otros. Con callbacks y temporizadores, este buscador acaba siendo una maraña de variables timeoutId, banderas y comprobaciones de “¿esta respuesta sigue siendo la última?”.

RxJS (Reactive Extensions for JavaScript) trata esos sucesos como una colección que se recorre en el tiempo. Igual que un array tiene map y filter, un flujo de eventos tiene map, filter y operadores que saben del tiempo y de la cancelación. El buscador de arriba se reduce a una lista de pasos, y al final del módulo lo tendrás funcionando:

texto$.pipe(
  map((texto) => texto.trim()),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap((texto) => this.buscar(texto)),
);

Figura 1. Cada operador resuelve una parte del problema. En la prueba real de este módulo, escribir “gar” produjo tres teclas y una sola petición.

Dónde te lo encuentras en Angular

No puedes esquivar RxJS del todo en Angular, porque varias APIs del propio framework devuelven observables:

API	Qué emite	Módulo
HttpClient.get() y similares	La respuesta y fin	12
Router.events	Eventos de navegación	11
ActivatedRoute.params	Parámetros de la ruta	10
FormControl.valueChanges	Cada cambio de un campo	17
rxResource	Usa un observable como cargador	13

¿Señales u observables?

Las señales no sustituyen a RxJS: resuelven otro problema. Esta tabla te orienta. El módulo 15 las une con toSignal y toObservable.

Pregunta	Señal	Observable
¿Qué representa?	Un estado con valor actual	Una secuencia de sucesos
¿Tiene valor siempre?	Sí	No necesariamente
¿Cómo se lee?	contador()	Hay que suscribirse
¿Cuándo se ejecuta?	computed calcula al leerse	El productor arranca al suscribirse
¿Sabe del tiempo?	No	Sí: esperas, ventanas, cancelación
¿Hay que darse de baja?	No	A veces (sección 12)
Uso típico	Estado que pinta la plantilla	Eventos, peticiones encadenadas, coordinación

💡 Consejo: Regla práctica: estado en señales, sucesos en observables. Si te preguntas “¿cuánto vale?”, es una señal. Si te preguntas “¿qué hago cada vez que pase esto?”, y además importa el orden o la cancelación, es un observable.

Cómo ejecutar los ejemplos

RxJS no necesita navegador ni Angular. La mayoría de los ejemplos del módulo son scripts que puedes lanzar desde la raíz de cualquier proyecto Angular, donde rxjs ya está instalado. Crea una carpeta pruebas/ junto a src/ y guarda cada ejemplo con extensión .mts:

node pruebas/primero.mts

Se han probado con Node 24, que ejecuta TypeScript borrando las anotaciones de tipos. La extensión .mts le dice a Node que el fichero es un módulo ES y evita un aviso.

⚠️ Importante: Como Node solo borra los tipos y no compila, un tipo importado debe llevar type: import { map, type OperatorFunction } from 'rxjs'. Sin esa palabra verás Named export 'OperatorFunction' not found, porque Node busca en tiempo de ejecución algo que solo existe para el compilador. En un proyecto Angular no ocurre, porque ahí compila TypeScript.

Por convenio, las variables que contienen un observable terminan en $, como texto$. No es obligatorio, pero ayuda a ver qué necesita suscripción.

---

2. Observable, Observer y Subscription

RxJS se apoya en tres piezas:

• Observable: la descripción de cómo se producen unos valores. Es una función guardada, una receta.
• Observer: el objeto que recibe esos valores, con tres métodos opcionales: next, error y complete.
• Subscription: lo que devuelve subscribe. Representa una ejecución en curso y permite cortarla.

Figura 2. Las tres piezas y el contrato que las une.

Tu primer Observable

// pruebas/primero.mts
import { Observable } from 'rxjs';

const numeros$ = new Observable<number>((suscriptor) => {
  console.log('Arranca el productor');
  suscriptor.next(1);
  suscriptor.next(2);

  const id = setTimeout(() => {
    suscriptor.next(3);
    suscriptor.complete();
  }, 100);

  // Lo que devuelve esta función se ejecuta al terminar la suscripción
  return () => {
    clearTimeout(id);
    console.log('Limpieza');
  };
});

console.log('Antes de suscribirse');
numeros$.subscribe({
  next: (valor) => console.log('Valor:', valor),
  error: (err) => console.error('Error:', err),
  complete: () => console.log('Completado'),
});
console.log('Después de suscribirse');

Antes de suscribirse
Arranca el productor
Valor: 1
Valor: 2
Después de suscribirse
Valor: 3
Completado
Limpieza

Esta salida enseña tres cosas que conviene grabarse:

1. Es perezoso. Crear numeros$ no ejecuta nada. “Arranca el productor” aparece solo tras subscribe.
2. Puede ser síncrono. Los valores 1 y 2 llegan antes de “Después de suscribirse”. Un observable no es asíncrono por naturaleza: lo es solo si su productor lo es.
3. Limpia al terminar. La función devuelta se ejecuta tras complete, tras error y tras unsubscribe.

El contrato

Una ejecución sigue siempre esta gramática: cero o más next, y como mucho un error o un complete. Tras un error o un complete, la ejecución ha terminado y RxJS ignora lo que venga después, aunque el productor insista:

const roto$ = new Observable<number>((s) => {
  s.next(1);
  s.error(new Error('Se acabó'));
  s.next(2);
  s.complete();
  return () => console.log('Limpieza tras el error');
});

roto$.subscribe({
  next: (v) => console.log('Valor:', v),
  error: (e: Error) => console.log('Error:', e.message),
  complete: () => console.log('Completado'),
});

Valor: 1
Error: Se acabó
Limpieza tras el error

El 2 no aparece y “Completado” tampoco. Esta regla explica buena parte de lo que verás en la sección 9: un error mata el flujo.

subscribe también admite una sola función, que hace de next: numeros$.subscribe((v) => console.log(v)). La forma con tres funciones sueltas, subscribe(fnNext, fnError, fnComplete), está obsoleta en RxJS 7. Si necesitas tratar errores, pasa un objeto.

Cortar con Subscription

Algunos flujos no terminan nunca, como un temporizador. Para detenerlos llamas a unsubscribe(), que ejecuta la limpieza:

const cuenta$ = new Observable<number>((suscriptor) => {
  let n = 0;
  const id = setInterval(() => suscriptor.next(++n), 100);
  console.log('Temporizador creado');

  return () => {
    clearInterval(id);
    console.log('Temporizador borrado');
  };
});

const suscripcion = cuenta$.subscribe((n) => {
  console.log('Tic', n);
  if (n === 3) {
    suscripcion.unsubscribe();
    console.log('closed:', suscripcion.closed);
  }
});

Temporizador creado
Tic 1
Tic 2
Tic 3
Temporizador borrado
closed: true

Sin la función de limpieza, el setInterval seguiría vivo aunque nadie escuchara. Por eso, cuando escribas un new Observable, pregúntate siempre qué recursos abre y devuelve la función que los cierra.

Puedes agrupar varias suscripciones en una sola con add. Al dar de baja la principal, se van todas:

const todas = new Subscription();
todas.add(interval(100).subscribe((n) => console.log('A', n)));
todas.add(interval(150).subscribe((n) => console.log('B', n)));
todas.add(() => console.log('Limpieza propia'));

setTimeout(() => todas.unsubscribe(), 320);

Observable frente a promesa

Si has trabajado con promesas (repásalas en el Curso de JavaScript si hace falta), la diferencia clave se ve al ejecutar esto:

const promesa = new Promise<string>((resolver) => {
  console.log('Promesa: el trabajo arranca sola');
  resolver('hecho');
});

const observable$ = new Observable<string>((s) => {
  console.log('Observable: el trabajo arranca al suscribirse');
  s.next('hecho');
  s.complete();
});

console.log('Fin de las declaraciones');
promesa.then((v) => console.log('then:', v));
observable$.subscribe((v) => console.log('subscribe:', v));
console.log('Fin del código síncrono');

Promesa: el trabajo arranca sola
Fin de las declaraciones
Observable: el trabajo arranca al suscribirse
subscribe: hecho
Fin del código síncrono
then: hecho

Característica	Promesa	Observable
Arranque	Inmediato, al crearla	Al suscribirse
Valores	Uno	Cero, uno o infinitos
Entrega	Siempre asíncrona	Síncrona o asíncrona
Cancelación	No tiene (AbortController aparte)	unsubscribe()
Varios consumidores	Comparten el mismo resultado	Cada uno lanza su ejecución (sección 11)
Operadores	then, catch	Más de cien

---

3. Diagramas de canicas

Razonar sobre el tiempo leyendo código cuesta. Por eso la comunidad de RxJS usa diagramas de canicas (marble diagrams): una línea es el tiempo, y cada canica es un valor emitido.

Figura 3. La entrada arriba, el operador en medio, la salida abajo. Una barra vertical significa que el flujo ha terminado bien; un aspa, que ha terminado con un error.

Los diagramas de este módulo se han dibujado a partir de ejecuciones reales. En la documentación oficial de RxJS verás los mismos dibujos para cada operador. Merece la pena leerlos antes que el texto.

También existe una versión en texto, muy cómoda para hacer pruebas (sección 15). Cada carácter es una unidad de tiempo: - no pasa nada, una letra es un valor, | es el fin y # es un error:

entrada:  -1-2-3-4-|
salida:   ---2---4-|

---

4. Crear observables

Casi nunca escribirás new Observable. RxJS trae funciones de creación para los casos habituales:

Función	Emite	Termina
of(1, 2, 3)	Sus argumentos, síncrono	Sí
from(array)	Cada elemento	Sí
from(promesa)	El resultado de la promesa	Sí
range(5, 3)	5, 6 y 7	Sí
timer(50)	Un 0 tras 50 ms	Sí
timer(0, 30)	0, 1, 2… empezando ya, cada 30 ms	No
interval(40)	0, 1, 2… cada 40 ms; el primero tras 40 ms	No
fromEvent(objetivo, 'click')	Cada evento	No
EMPTY	Nada	Sí, en el acto
NEVER	Nada	No
throwError(() => error)	Un error	Con error
defer(() => obs$)	Lo que dé la función, creada al suscribirse	Depende

Un ejemplo con varias, usando un observador que indica qué pasa:

import { EMPTY, from, of, range, throwError } from 'rxjs';

const observador = (nombre: string) => ({
  next: (v: unknown) => console.log(`${nombre}:`, v),
  error: (e: Error) => console.log(`${nombre}: error "${e.message}"`),
  complete: () => console.log(`${nombre}: fin`),
});

of(1, 2).subscribe(observador('of'));
from(['Ventas', 'I+D']).subscribe(observador('from array'));
range(5, 3).subscribe(observador('range'));
EMPTY.subscribe(observador('EMPTY'));
throwError(() => new Error('falló')).subscribe(observador('throwError'));

of: 1
of: 2
of: fin
from array: Ventas
from array: I+D
from array: fin
range: 5
range: 6
range: 7
range: fin
EMPTY: fin
throwError: error "falló"

from convierte casi todo

Muchos operadores aceptan un ObservableInput, es decir, cualquier cosa que from sepa convertir: un observable, una promesa, un array o un iterable. Por eso switchMap admite que tu función devuelva una promesa.

fromEvent y sus límites en Angular

fromEvent escucha eventos de cualquier EventTarget, y unsubscribe() quita el listener:

const objetivo = new EventTarget();
const clics$ = fromEvent(objetivo, 'clic');

const baja = clics$.subscribe((e) => console.log('Evento:', e.type));
objetivo.dispatchEvent(new Event('clic'));   // Evento: clic
baja.unsubscribe();
objetivo.dispatchEvent(new Event('clic'));   // no imprime nada

En un componente Angular no sueles necesitarlo. Es más limpio escuchar con (input) o (click) en la plantilla y reenviar el evento a un Subject, como harás en la sección 7. Así no hay que buscar el elemento en el DOM.

defer: crear en el último momento

of(Date.now()) calcula la hora al declarar, una sola vez. defer retrasa la creación hasta cada suscripción:

const ahoraDirecto$ = of(Date.now());
const ahoraDiferido$ = defer(() => of(Date.now()));

Tras esperar 100 ms y suscribirse a cada uno, la comprobación da:

of ya lleva 100 ms de antigüedad
defer lleva 0 ms de antigüedad

defer es la forma correcta de envolver algo que arranca solo, como una promesa. from(fetch(url)) lanza la petición al declarar, y reintentar no servirá de nada porque la promesa ya está hecha. defer(() => fetch(url)) la lanza en cada suscripción.

⚠️ Importante: throwError recibe una función que crea el error: throwError(() => new Error('x')). La forma con el valor directo, throwError(new Error('x')), está obsoleta en RxJS 7.

---

5. Operadores básicos

Un operador es una función que recibe un observable y devuelve otro nuevo. No modifica el original. Se encadenan con pipe:

import { filter, from, map } from 'rxjs';

const plantilla = [
  { nombre: 'Ana', salario: 3200, departamento: 10 },
  { nombre: 'Luis', salario: 2100, departamento: 20 },
  { nombre: 'Marta', salario: 2900, departamento: 10 },
  { nombre: 'Jorge', salario: 1800, departamento: 30 },
];

from(plantilla).pipe(
  filter((e) => e.departamento === 10),
  map((e) => `${e.nombre}: ${e.salario} €`),
).subscribe(console.log);

Ana: 3200 €
Marta: 2900 €

Figura 4. Cada operador se suscribe al anterior y crea un observable nuevo.

Un valor recorre toda la tubería antes del siguiente

Parece que primero se filtra todo y luego se transforma, como con los arrays. No es así. Cada valor baja entero por la tubería, y hasta que no termina no entra el siguiente. tap permite verlo, porque ejecuta una acción sin alterar el valor:

of(1, 2, 3, 4).pipe(
  tap((n) => console.log('entra', n)),
  filter((n) => n % 2 === 0),
  tap((n) => console.log('  pasa el filtro', n)),
  map((n) => n * 10),
).subscribe((n) => console.log('    llega', n));

entra 1
entra 2
  pasa el filtro 2
    llega 20
entra 3
entra 4
  pasa el filtro 4
    llega 40

💡 Consejo: tap es tu herramienta de depuración. Coloca tap((v) => console.log('aquí', v)) entre dos operadores para ver qué pasa por ese punto. También se usa para efectos secundarios, como activar un indicador de carga.

Acumular: scan y reduce

reduce funciona como en los arrays: da un único resultado al terminar el flujo. scan hace lo mismo, pero emite cada resultado intermedio. En un flujo que no termina, reduce no emitirá nunca, y por eso scan es el habitual en interfaces:

const salarios$ = from(plantilla).pipe(map((e) => e.salario));

salarios$.pipe(scan((total, s) => total + s, 0)).subscribe((t) => console.log('scan', t));
salarios$.pipe(reduce((total, s) => total + s, 0)).subscribe((t) => console.log('reduce', t));

scan 3200
scan 5300
scan 8200
scan 10000
reduce 10000

Seleccionar valores

of(1, 2, 3, 4, 5).pipe(take(2));                    // 1, 2
of(1, 2, 3, 4, 5).pipe(skip(3));                    // 4, 5
of(1, 2, 3, 4, 5).pipe(first());                    // 1
of(1, 2, 3, 4, 5).pipe(first((n) => n > 3));        // 4
of(1, 2, 3, 4, 5).pipe(last());                     // 5
of(1, 2, 3, 4, 5).pipe(takeWhile((n) => n < 3));    // 1, 2
of('a', 'a', 'b', 'a', 'a').pipe(distinctUntilChanged());   // a, b, a
of(1, 2, 3, 4, 5).pipe(bufferCount(2));             // [1, 2], [3, 4], [5]
of(10, 12, 9).pipe(pairwise());                     // [10, 12], [12, 9]
of('hola').pipe(startWith('primero'));              // primero, hola

Fíjate en distinctUntilChanged: elimina repeticiones consecutivas, no todas. Por eso la última a sobrevive tras la b. Por defecto compara con ===. Con objetos, dos elementos iguales de contenido son distintos para ===, así que indica tú la comparación:

of({ id: 1, n: 'x' }, { id: 1, n: 'y' }, { id: 2, n: 'z' }).pipe(
  distinctUntilChanged((a, b) => a.id === b.id),
);   // x, z

Operador	Qué hace	Equivalente en arrays
map(f)	Transforma cada valor	map
filter(f)	Deja pasar los que cumplen	filter
tap(f)	Efecto secundario, sin cambiar el valor	Ninguno
reduce(f, ini)	Un resultado al final	reduce
scan(f, ini)	Cada resultado intermedio	Ninguno
take(n)	Los n primeros y termina	slice(0, n)
skip(n)	Salta los n primeros	slice(n)
first(), last()	El primero o el último	at(0), at(-1)
takeWhile(f)	Hasta que la condición falle	Ninguno
distinctUntilChanged()	Quita repetidos consecutivos	Ninguno
startWith(v)	Antepone valores	unshift
pairwise()	Pares [anterior, actual]	Ninguno

---

6. Operadores de aplanamiento

Esta es la parte de RxJS que más dudas genera, y la que más usarás. Empieza con un problema real: cargar un empleado y, con su deptId, cargar su departamento.

La primera idea suele ser anidar suscripciones:

// MAL: suscripciones anidadas
this.http.get<Empleado>(url).subscribe((empleado) => {
  this.http.get<Departamento>(`${API}/departamentos/${empleado.deptId}`).subscribe((departamento) => {
    this.datos.set({ empleado, departamento });
  });
});

Funciona, pero tiene tres problemas: unsubscribe no alcanza a la petición interna, cada nivel necesita su tratamiento de errores, y si esto se repite por cada clic nadie decide qué ocurre con las peticiones que siguen en marcha.

Y con map tampoco sale, porque cada valor se convierte en otro observable, y obtienes un observable de observables que nadie ejecuta:

empleado$.pipe(
  map((e) => this.http.get<Departamento>(`${API}/departamentos/${e.deptId}`)),
).subscribe((d) => console.log(d));   // imprime un Observable, sin hacer ninguna petición

Un operador de aplanamiento resuelve las dos cosas. Recibe una función que devuelve un observable interno, se suscribe a él y envía sus valores a la salida. La única diferencia entre los cuatro operadores es qué hacen cuando llega un valor nuevo y aún hay un observable interno en marcha.

Los cuatro, con el mismo experimento

Tres clics en 0, 100 y 200 ms. Cada uno lanza una petición simulada: A tarda 350 ms, B 150 ms y C 100 ms. Este script usa un reloj virtual de RxJS, que verás en la sección 15. Gracias a él, los tiempos de la salida son exactos y no hace falta esperar:

// pruebas/aplanadores.mts
import { concatMap, exhaustMap, map, merge, mergeMap, switchMap, tap, timer,
  type Observable, type OperatorFunction } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';

function virtual(fn: (ahora: () => number) => void) {
  const planificador = new TestScheduler(() => undefined);
  planificador.run(() => fn(() => planificador.now()));
}

function probar(
  nombre: string,
  aplanador: (f: (l: string) => Observable<string>) => OperatorFunction<string, string>,
) {
  console.log('--- ' + nombre);
  virtual((ahora) => {
    const reg = (m: string) => console.log(String(ahora()).padStart(4), 'ms', m);
    const duracion: Record<string, number> = { A: 350, B: 150, C: 100 };

    const peticion = (letra: string) =>
      timer(duracion[letra]).pipe(
        map(() => letra),
        tap({
          subscribe: () => reg(`empieza ${letra}`),
          unsubscribe: () => reg(`CANCELADA ${letra}`),
        }),
      );

    const clics$ = merge(
      timer(0).pipe(map(() => 'A')),
      timer(100).pipe(map(() => 'B')),
      timer(200).pipe(map(() => 'C')),
    );

    clics$.pipe(tap((l) => reg(`clic ${l}`)), aplanador(peticion)).subscribe({
      next: (r) => reg(`resultado ${r}`),
      complete: () => reg('fin'),
    });
  });
}

probar('mergeMap', (f) => mergeMap(f));
probar('concatMap', (f) => concatMap(f));
probar('switchMap', (f) => switchMap(f));
probar('exhaustMap', (f) => exhaustMap(f));

--- mergeMap
   0 ms clic A
   0 ms empieza A
 100 ms clic B
 100 ms empieza B
 200 ms clic C
 200 ms empieza C
 250 ms resultado B
 300 ms resultado C
 350 ms resultado A
 350 ms fin
--- concatMap
   0 ms clic A
   0 ms empieza A
 100 ms clic B
 200 ms clic C
 350 ms resultado A
 350 ms empieza B
 500 ms resultado B
 500 ms empieza C
 600 ms resultado C
 600 ms fin
--- switchMap
   0 ms clic A
   0 ms empieza A
 100 ms clic B
 100 ms CANCELADA A
 100 ms empieza B
 200 ms clic C
 200 ms CANCELADA B
 200 ms empieza C
 300 ms resultado C
 300 ms fin
--- exhaustMap
   0 ms clic A
   0 ms empieza A
 100 ms clic B
 200 ms clic C
 350 ms resultado A
 350 ms fin

Figura 5. El mismo experimento en un diagrama. Mira la columna de 300 ms: cada operador ha tomado una decisión distinta con las mismas entradas.

Puedes recordarlos con una situación cotidiana. Piensa en un camarero que recibe pedidos:

• mergeMap (todas a la vez): atiende a todos en paralelo. Los platos salen según se terminan, sin respetar el orden de llegada.
• concatMap (en cola): termina un pedido antes de empezar el siguiente. Respeta el orden, pero hace esperar.
• switchMap (la última gana): si llega un pedido nuevo, tira el que estaba preparando. Solo importa el último.
• exhaustMap (ocupado): mientras prepara uno, no escucha a nadie. Lo que llegue mientras tanto se pierde.

El problema de la carrera, y por qué switchMap lo evita

En un buscador, cada tecla lanza una petición, y el servidor no promete contestar en orden. Si la búsqueda de “a” devuelve muchas filas y tarda 300 ms, y la de “an” tarda 100 ms, ¿qué ve el usuario?

const buscar = (texto: string) =>
  timer(texto === 'a' ? 300 : 100).pipe(map(() => `resultados de "${texto}"`));

const teclas = () => merge(
  timer(0).pipe(map(() => 'a')),
  timer(50).pipe(map(() => 'an')),
);

teclas().pipe(mergeMap(buscar)).subscribe((r) => console.log('mergeMap pinta', r));
// ... y después, por separado:
teclas().pipe(switchMap(buscar)).subscribe((r) => console.log('switchMap pinta', r));

mergeMap pinta resultados de "an"
mergeMap pinta resultados de "a"
switchMap pinta resultados de "an"

Con mergeMap la respuesta antigua llega la última y pisa a la buena: el cuadro de búsqueda dice “an” y la lista muestra los resultados de “a”. Este error se llama condición de carrera y solo aparece a veces, según la red. switchMap lo elimina por construcción, porque al llegar “an” se da de baja de “a”.

Cancelar de verdad: qué pasa con la petición 🔄 v22.0

🔄 Cambia en Angular 22.0. Darse de baja de HttpClient aborta la petición en las dos versiones, pero en la LTS (Angular 21) el transporte por defecto es XHR y cancela con xhr.abort(), mientras que en 22 es fetch. Si usas la LTS: no tienes que cambiar nada en tu código; en la pestaña Red verás el tipo xhr en lugar de fetch.

En RxJS, cancelar significa darse de baja. Con HttpClient, esa baja se traduce en una cancelación real en el navegador. En este módulo se ha comprobado con la ficha de empleado de la sección 9, cambiando el identificador a 101, 102 y 103 seguidos con switchMap y un servidor algo lento. Las dos primeras peticiones aparecieron como fallidas en el navegador:

/empleados/101 net::ERR_ABORTED fetch
/empleados/102 net::ERR_ABORTED fetch

Eso ahorra ancho de banda en el cliente, pero no garantiza que el servidor no haya hecho el trabajo. Puede que la petición ya haya llegado.

⚠️ Importante: Por eso switchMap es peligroso para escrituras. Si cancelas un POST a mitad, no sabes si el servidor lo procesó. Para guardar usa concatMap o exhaustMap.

Limitar la concurrencia de mergeMap

mergeMap admite como segundo argumento cuántos observables internos ejecutar a la vez. Es muy útil para no lanzar cien peticiones de golpe. Con un máximo de 2 y cinco identificadores:

from([1, 2, 3, 4, 5]).pipe(
  mergeMap((id) => {
    console.log('pide', id);
    return timer(100).pipe(map(() => id));
  }, 2),
).subscribe((id) => console.log('recibe', id));

pide 1
pide 2
recibe 1
pide 3
recibe 2
pide 4
recibe 3
pide 5
recibe 4
recibe 5

concatMap equivale a mergeMap con concurrencia 1.

Cómo elegir

Operador	Con un valor nuevo	Orden de salida	Caso típico
switchMap	Cancela la interna anterior	El del último	Buscador, filtros, cambio de ruta: lecturas
concatMap	Lo pone en cola	Se conserva	Guardados que deben aplicarse en orden
mergeMap	Lanza otra en paralelo	Según terminen	Tareas independientes, como descargas
exhaustMap	Lo ignora	El del primero	Botón de enviar, inicio de sesión

💡 Consejo: Si dudas, hazte dos preguntas. ¿Sirve de algo el resultado anterior si llega uno nuevo? Si no, switchMap. ¿Importa el orden? Si sí, concatMap. ¿Quieres evitar el doble clic? exhaustMap. Si son tareas independientes, mergeMap.

---

7. Operadores de tiempo y el buscador completo

Hay eventos que llegan demasiado deprisa: teclas, movimientos del ratón, desplazamiento de página. Cuatro operadores regulan ese caudal con una ventana de tiempo, y cada uno hace una cosa diferente:

// pruebas/tiempo.mts: reutiliza la función virtual de la sección 6
const operadores: [string, MonoTypeOperatorFunction<string>][] = [
  ['debounceTime', debounceTime<string>(300)],
  ['throttleTime', throttleTime<string>(300)],
  ['auditTime', auditTime<string>(300)],
  ['sampleTime', sampleTime<string>(300)],
];

for (const [nombre, operador] of operadores) {
  virtual((ahora) => {
    const teclas: [number, string][] = [
      [0, 'a'], [100, 'b'], [200, 'c'], [700, 'd'], [800, 'e'], [1300, 'f'],
    ];
    const tecleo$ = merge(...teclas.map(([ms, v]) => timer(ms).pipe(map(() => v))));
    const salida: string[] = [];

    tecleo$.pipe(operador).subscribe({
      next: (v) => salida.push(`${v}@${ahora()}`),
      complete: () => console.log(nombre.padEnd(13), salida.join('  '), ` (fin en ${ahora()})`),
    });
  });
}

debounceTime  c@500  e@1100  f@1300  (fin en 1300)
throttleTime  a@0  d@700  f@1300  (fin en 1300)
auditTime     c@300  e@1000  f@1600  (fin en 1600)
sampleTime    c@300  e@900  (fin en 1300)

Figura 6. La salida real de los cuatro operadores, medida con reloj virtual.

• debounceTime(300): espera a que pasen 300 ms sin eventos y emite el último. Si llega otro, reinicia la espera. Fíjate en que el flujo acaba en 1300 y emite f en ese instante: al completarse, entrega el valor pendiente.
• throttleTime(300): emite el primero y cierra la puerta durante 300 ms. Da respuesta inmediata.
• auditTime(300): al llegar un evento abre una ventana de 300 ms, y al cerrarla emite el último.
• sampleTime(300): mira el reloj cada 300 ms y emite el último valor si ha llegado uno nuevo. Aquí f se pierde, porque el flujo termina antes de la siguiente muestra.

Operador	Emite	Uso típico
debounceTime	Cuando se calma	Buscador, autoguardado, validación al escribir
throttleTime	Al principio de cada ventana	Evitar repeticiones en un botón, disparos en un juego
auditTime	Al final de cada ventana	Redimensionado y desplazamiento
sampleTime	A intervalos fijos	Refrescar un gráfico a ritmo constante

El buscador de EMPRESA

Con lo visto, ya puedes montar el buscador. Necesitas la API EMPRESA en marcha: descarga empresa-api.zip, descomprímelo y ejecuta node server.mjs.

Primero, un fichero con los tipos que usarán los ejemplos Angular del módulo:

// src/app/modelos.ts
export const API = 'http://localhost:3000/api';

export interface Pagina<T> {
  contenido: T[];
  pagina: number;
  tamano: number;
  total: number;
  paginas: number;
}

export interface Empleado {
  empId: number;
  nombre: string;
  apellidos: string;
  oficio: string;
  salario: number;
  deptId: number;
  departamento: string;
}

export interface Departamento {
  deptId: number;
  nombre: string;
  localidad: string;
  presupuesto: number;
}

export interface Producto {
  productoId: number;
  nombre: string;
  precio: number;
  stock: number;
}

export interface Proyecto {
  proyectoId: number;
  nombre: string;
}

export interface Estadisticas {
  empleados: number;
  departamentos: number;
  productos: number;
  pedidos: number;
  facturacion: number;
}

Y el componente:

// src/app/buscador.ts
import { HttpClient, HttpParams } from '@angular/common/http';
import { Component, DestroyRef, inject, signal } from '@angular/core';
import {
  Observable, Subject, catchError, debounceTime, distinctUntilChanged,
  map, of, switchMap, takeUntil, tap,
} from 'rxjs';
import { API, Empleado, Pagina } from './modelos';

@Component({
  selector: 'app-buscador',
  template: `
    <label>
      Buscar empleado
      <input type="search" placeholder="Nombre o apellidos" (input)="escribir($event)" />
    </label>

    @if (cargando()) {
      <p class="aviso">Buscando...</p>
    }
    @if (error(); as mensaje) {
      <p class="error">{{ mensaje }}</p>
    }

    <ul>
      @for (e of empleados(); track e.empId) {
        <li>{{ e.nombre }} {{ e.apellidos }} <small>{{ e.departamento }}</small></li>
      } @empty {
        <li>Sin resultados</li>
      }
    </ul>
  `,
})
export class Buscador {
  private readonly http = inject(HttpClient);
  private readonly destruido$ = new Subject<void>();
  private readonly texto$ = new Subject<string>();

  protected readonly empleados = signal<Empleado[]>([]);
  protected readonly cargando = signal(false);
  protected readonly error = signal<string | null>(null);

  constructor() {
    inject(DestroyRef).onDestroy(() => {
      this.destruido$.next();
      this.destruido$.complete();
    });

    this.texto$
      .pipe(
        map((texto) => texto.trim()),
        debounceTime(300),
        distinctUntilChanged(),
        tap(() => {
          this.cargando.set(true);
          this.error.set(null);
        }),
        switchMap((texto) => this.buscar(texto)),
        takeUntil(this.destruido$),
      )
      .subscribe((lista) => {
        this.empleados.set(lista);
        this.cargando.set(false);
      });
  }

  protected escribir(evento: Event) {
    this.texto$.next((evento.target as HTMLInputElement).value);
  }

  private buscar(texto: string): Observable<Empleado[]> {
    const params = new HttpParams().set('q', texto).set('tamano', 6);
    return this.http.get<Pagina<Empleado>>(`${API}/empleados`, { params }).pipe(
      map((pagina) => pagina.contenido),
      catchError(() => {
        this.error.set('No se ha podido completar la búsqueda');
        return of([]);
      }),
    );
  }
}

Figura 7. El buscador en marcha. La API encuentra “gar” en “García” sin distinguir mayúsculas ni tildes.

Lee la tubería de arriba abajo, porque cada línea tiene un motivo:

• texto$ es un Subject. Es un observable en el que puedes escribir con next (sección 10). Hace de puente entre el evento de la plantilla y el mundo RxJS.
• map antes de debounceTime limpia el texto, de modo que “ana” y “ana “ cuentan como iguales.
• distinctUntilChanged después de debounceTime. Si el usuario escribe una letra y la borra enseguida, el texto calmado no ha cambiado, y no merece otra petición. Si lo pones antes, cada tecla sería distinta y no serviría de nada.
• catchError dentro de buscar. Si estuviera fuera, el primer fallo mataría el flujo y el buscador dejaría de responder (sección 9).
• takeUntil el último. Corta todo al destruir el componente (sección 12).
• Los resultados van a señales. Las plantillas se actualizan al cambiar una señal o al emitir async. Con una aplicación zoneless (módulo 09), asignar a una propiedad normal dentro de subscribe no repinta la vista.

📘 Concepto: Hay un fallo sutil en este código, que aparece con frecuencia en proyectos reales. Si el usuario escribe un texto, lo borra y vuelve al mismo, no hay problema. Pero si la búsqueda falla y repite el mismo texto, distinctUntilChanged no le deja reintentar. En un buscador rara vez importa. Si te importa, añade un botón de reintentar que envíe directamente a buscar.

El módulo 15 simplifica este componente: toSignal sustituye a la suscripción manual y takeUntilDestroyed sustituye a destruido$. Antes de usar esos atajos conviene entender lo que hay debajo.

---

8. Combinar flujos

Cuando una pantalla depende de varias fuentes, tienes que decidir qué dispara la salida. Esa es la diferencia entre los operadores de esta sección.

Figura 8. Mismas entradas, cinco resultados. En la salida, b1 significa “b de la entrada a, 1 de la entrada b”.

Con dos entradas temporizadas, estas son las salidas reales:

// dep emite en 0 y 200; orden emite en 100 y 300
combineLatest([dep$, orden$]).subscribe(([d, o]) => console.log('combineLatest', d, o));

combineLatest dep10 nombre
combineLatest dep20 nombre
combineLatest dep20 salario

// la principal emite a, b y c en 0, 100 y 200; la otra, x e y en 50 y 150
principal$.pipe(withLatestFrom(otra$)).subscribe((p) => console.log('withLatestFrom', p));

withLatestFrom [ 'b', 'x' ]
withLatestFrom [ 'c', 'y' ]

Con withLatestFrom, a se pierde porque en ese momento la otra entrada no ha emitido nada. Lo mismo ocurre en combineLatest: no emite hasta que todas sus entradas hayan emitido una vez. Si una no llega nunca, no sale nada. Por eso es habitual dar un valor inicial con startWith o partir de un BehaviorSubject.

Función	Emite cuando…	Termina cuando…	Uso típico
combineLatest	Cualquiera emite, una vez que todas lo han hecho	Terminan todas	Filtros combinados
withLatestFrom	Emite la principal	Termina la principal	Un clic que necesita el estado actual
forkJoin	Terminan todas, una sola vez	Igual	Peticiones en paralelo
zip	Hay un valor nuevo de cada una	Una se agota	Emparejar por posición
merge	Cualquiera emite	Terminan todas	Unir varias fuentes de eventos
concat	Una tras otra	Termina la última	Secuencia de pasos
race	Gana la primera que emita	Termina la ganadora	Elegir la fuente más rápida

concat(of('primero'), timer(50).pipe(map(() => 'segundo')), of('tercero'));
// primero, segundo, tercero

race(
  timer(100).pipe(map(() => 'lento')),
  timer(30).pipe(map(() => 'rápido')),
);
// rápido

Filtros con combineLatest

Un listado con dos filtros es el caso de libro. Cada filtro es un BehaviorSubject, que siempre tiene valor, de modo que combineLatest arranca enseguida. switchMap lanza la petición:

// src/app/listado.ts
import { AsyncPipe, CurrencyPipe } from '@angular/common';
import { HttpClient, HttpParams } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { BehaviorSubject, combineLatest, map, switchMap } from 'rxjs';
import { API, Departamento, Empleado, Pagina } from './modelos';

@Component({
  selector: 'app-listado',
  imports: [AsyncPipe, CurrencyPipe],
  template: `
    <label>
      Departamento
      <select #dep (change)="deptId$.next(dep.value ? +dep.value : null)">
        <option value="">Todos</option>
        @for (d of departamentos$ | async; track d.deptId) {
          <option [value]="d.deptId">{{ d.nombre }}</option>
        }
      </select>
    </label>
    <label>
      Ordenar por
      <select #ord (change)="orden$.next(ord.value)">
        <option value="apellidos">Apellidos</option>
        <option value="-salario">Salario descendente</option>
      </select>
    </label>

    <table>
      <tbody>
        @for (e of empleados$ | async; track e.empId) {
          <tr>
            <td>{{ e.nombre }} {{ e.apellidos }}</td>
            <td>{{ e.salario | currency: 'EUR' }}</td>
          </tr>
        }
      </tbody>
    </table>
  `,
})
export class Listado {
  private readonly http = inject(HttpClient);

  protected readonly deptId$ = new BehaviorSubject<number | null>(null);
  protected readonly orden$ = new BehaviorSubject<string>('apellidos');

  protected readonly departamentos$ = this.http
    .get<Pagina<Departamento>>(`${API}/departamentos`)
    .pipe(map((pagina) => pagina.contenido));

  protected readonly empleados$ = combineLatest([this.deptId$, this.orden$]).pipe(
    switchMap(([deptId, orden]) => {
      let params = new HttpParams().set('orden', orden).set('tamano', 5);
      if (deptId !== null) {
        params = params.set('deptId', deptId);
      }
      return this.http.get<Pagina<Empleado>>(`${API}/empleados`, { params });
    }),
    map((pagina) => pagina.contenido),
  );
}

Figura 9. Cambiar cualquier filtro lanza una petición con los valores actuales de ambos. La moneda sale con el formato regional por defecto; en el módulo 07 viste cómo cambiarlo.

Aquí no hay subscribe ni baja manual: AsyncPipe se suscribe al entrar y se da de baja al destruirse el componente. Recuerda que cada | async es una suscripción propia. Si usaras empleados$ | async dos veces en la plantilla, habría dos peticiones por cada cambio. La sección 11 explica el motivo.

Peticiones en paralelo con forkJoin

forkJoin es el Promise.all de RxJS: espera a que terminen todas y da el último valor de cada una. Admite un objeto, y así obtienes nombres en lugar de posiciones. Lo verás funcionando en el ejercicio 4.

forkJoin({
  deps: of('x', 'departamentos').pipe(delay(100)),
  emps: of('empleados').pipe(delay(50)),
}).subscribe((r) => console.log(r));

{ deps: 'departamentos', emps: 'empleados' }

Tiene dos trampas, ambas consecuencia de su definición:

• Si un flujo no termina, nunca emite. No le pases valueChanges ni un interval.
• Si uno falla, falla todo y se pierden los resultados de los demás. Pon un catchError en cada entrada que pueda fallar sin arrastrar a las otras.

---

9. Errores

Un error en RxJS viaja por su propio canal. Baja por la tubería saltándose los map y filter hasta encontrar un catchError o llegar al error del observador. Una excepción lanzada dentro de un operador se convierte en una notificación de error:

const inestable$ = of(1, 2, 3).pipe(
  map((n) => {
    if (n === 2) throw new Error('El 2 no me gusta');
    return n * 10;
  }),
);

inestable$.pipe(
  catchError((err: Error) => {
    console.log('Capturado:', err.message);
    return of(-1);
  }),
).subscribe({
  next: (v) => console.log('Valor:', v),
  complete: () => console.log('Completado'),
});

Valor: 10
Capturado: El 2 no me gusta
Valor: -1
Completado

Observa que el 3 no llega. catchError no retoma el flujo donde estaba: lo sustituye por el observable que devuelves. El contrato de la sección 2 sigue vigente, y el flujo original ha muerto. Puedes devolver:

• of(valorPorDefecto) para continuar con un sustituto
• EMPTY para terminar sin emitir
• throwError(() => otroError) para relanzarlo, por ejemplo transformado

Dónde se coloca catchError

Es uno de los errores más frecuentes en aplicaciones reales. Compara estas dos versiones, en las que el identificador 2 falla:

const peticion = (id: number) =>
  id === 2 ? throwError(() => new Error('404')) : of(`empleado ${id}`);

// Fuera: captura en la tubería principal
from([1, 2, 3]).pipe(
  concatMap((id) => peticion(id)),
  catchError(() => of('sin datos')),
).subscribe({ next: (v) => console.log('Fuera:', v), complete: () => console.log('Fuera: fin') });

// Dentro: captura en cada petición
from([1, 2, 3]).pipe(
  concatMap((id) => peticion(id).pipe(catchError(() => of('sin datos')))),
).subscribe({ next: (v) => console.log('Dentro:', v), complete: () => console.log('Dentro: fin') });

Fuera: empleado 1
Fuera: sin datos
Fuera: fin
Dentro: empleado 1
Dentro: sin datos
Dentro: empleado 3
Dentro: fin

Figura 10. Si el flujo principal viene de clics, la versión “fuera” deja la pantalla muerta sin ningún aviso, y el usuario pulsa sin que ocurra nada.

⚠️ Importante: Regla: en un flujo que debe seguir vivo, el catchError va dentro de switchMap, mergeMap y similares, en el observable interno.

Modelar la carga como estados

Un patrón muy claro consiste en convertir el resultado de cada petición en un valor de un tipo unión. Así, cargando, éxito y error pasan a ser datos normales que la plantilla puede dibujar:

// src/app/ficha.ts
import { AsyncPipe, CurrencyPipe } from '@angular/common';
import { HttpClient, HttpErrorResponse } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import {
  BehaviorSubject, Observable, catchError, distinctUntilChanged,
  map, of, startWith, switchMap,
} from 'rxjs';
import { API, Empleado } from './modelos';

type Estado<T> =
  | { tipo: 'cargando' }
  | { tipo: 'ok'; datos: T }
  | { tipo: 'error'; mensaje: string };

@Component({
  selector: 'app-ficha',
  imports: [AsyncPipe, CurrencyPipe],
  template: `
    <label>
      Identificador
      <input #campo type="number" value="100" (change)="id$.next(campo.valueAsNumber)" />
    </label>

    @if (estado$ | async; as estado) {
      @switch (estado.tipo) {
        @case ('cargando') {
          <p class="aviso">Cargando...</p>
        }
        @case ('ok') {
          <p>
            <strong>{{ estado.datos.nombre }} {{ estado.datos.apellidos }}</strong><br />
            {{ estado.datos.oficio }}, {{ estado.datos.salario | currency: 'EUR' }}
          </p>
        }
        @case ('error') {
          <p class="error">{{ estado.mensaje }}</p>
        }
      }
    }
  `,
})
export class Ficha {
  private readonly http = inject(HttpClient);

  protected readonly id$ = new BehaviorSubject<number>(100);

  protected readonly estado$: Observable<Estado<Empleado>> = this.id$.pipe(
    distinctUntilChanged(),
    switchMap((id) =>
      this.http.get<Empleado>(`${API}/empleados/${id}`).pipe(
        map((datos): Estado<Empleado> => ({ tipo: 'ok', datos })),
        catchError((error: HttpErrorResponse) =>
          of<Estado<Empleado>>({ tipo: 'error', mensaje: error.error?.mensaje ?? 'Error de red' }),
        ),
        startWith<Estado<Empleado>>({ tipo: 'cargando' }),
      ),
    ),
  );
}

Fíjate en el orden de la interna: startWith va después de map y catchError, para que el estado “cargando” llegue a la salida sin pasar por el map. Tampoco hace falta ninguna señal cargando, porque el estado viaja en el propio flujo.

Figura 11. La ficha tras cambiar el identificador a un empleado que existe.

Figura 12. Con un identificador inexistente, la API responde 404 y su mensaje llega a la plantilla. Como el catchError está dentro, el campo sigue funcionando.

Reintentar con retry

retry se vuelve a suscribir al observable cuando falla. Con un observable frío, eso significa repetir el trabajo. Aquí defer permite contar los intentos:

let intentos = 0;
const fragil$ = defer(() => {
  intentos++;
  console.log('Intento', intentos);
  return intentos < 3 ? throwError(() => new Error('503')) : of('ya funciona');
});

fragil$.pipe(retry(2)).subscribe({
  next: console.log,
  error: (e: Error) => console.log('Falló:', e.message),
});

Intento 1
Intento 2
Intento 3
ya funciona

Reintentar sin espera contra un servidor saturado solo lo empeora. retry acepta una configuración con count, delay y resetOnSuccess. La función delay recibe el error y el número de intento, y devuelve un observable: cuando este emite, se reintenta. La espera exponencial duplica el tiempo cada vez:

fuente$.pipe(
  retry({
    count: 3,
    delay: (_error, intento) => timer(100 * 2 ** (intento - 1)),
  }),
);

0 ms: intento 1
100 ms: intento 2
300 ms: intento 3
800 ms: intento 4
Me rindo: 503

Las esperas son 100, 200 y 400 ms. retryWhen, que verás en código antiguo, está obsoleto en favor de esta opción delay. En el módulo 12 viste dónde reintentar peticiones HTTP. Recuerda que reintentar solo tiene sentido en errores pasajeros y en operaciones que se pueden repetir sin efectos: un 404 no se arregla por insistir. El ejercicio 5 lo resuelve.

timeout y finalize

timeout da un error si no llega un valor a tiempo. En su forma de objeto, permite usar otro flujo de repuesto:

const cuelga$ = timer(500).pipe(map(() => 'tarde'));

cuelga$.pipe(timeout(100)).subscribe({
  error: (e) => console.log(e instanceof TimeoutError, e.name, e.message),
});
cuelga$.pipe(
  timeout({ first: 100, with: () => of('valor de repuesto') }),
).subscribe((v) => console.log(v));

true TimeoutError Timeout has occurred
valor de repuesto

finalize ejecuta una función cuando la suscripción termina por cualquier motivo. Es el finally de RxJS, y el sitio correcto para quitar un indicador de carga:

of('dato').pipe(finalize(() => console.log('finalize: tras completar'))).subscribe();

throwError(() => new Error('x')).pipe(
  finalize(() => console.log('finalize: tras error')),
).subscribe({ error: () => console.log('error') });

interval(10).pipe(finalize(() => console.log('finalize: tras baja'))).subscribe().unsubscribe();

finalize: tras completar
error
finalize: tras error
finalize: tras baja

Si nadie atiende el error

Si no pasas un error al observador, RxJS relanza el error de forma asíncrona, para no romper el código que estaba en marcha:

throwError(() => new Error('nadie me escucha')).subscribe((v) => console.log(v));
console.log('El código sigue');

El código sigue
Error: nadie me escucha
    ...

En Node eso detiene el proceso. En el navegador lo verás como un error sin capturar en la consola.

---

10. Subjects

Un observable normal es de solo lectura: el productor está dentro. Un Subject es a la vez observable y observador. Tiene subscribe y también next, error y complete, y reenvía lo que recibe a todos sus suscriptores. Es la forma de pasar del código imperativo al mundo RxJS, como hiciste con texto$ en el buscador.

const avisos$ = new Subject<string>();

avisos$.subscribe((v) => console.log('Primero:', v));
avisos$.next('uno');

avisos$.subscribe((v) => console.log('Segundo:', v));
avisos$.next('dos');

avisos$.complete();
avisos$.next('tres');
avisos$.subscribe({ complete: () => console.log('Tardío: ya estaba completado') });

Primero: uno
Primero: dos
Segundo: dos
Tardío: ya estaba completado

El segundo suscriptor no recibe “uno”, porque llegó tarde. Un Subject no guarda memoria. Y tras complete ya no emite nada más. Hay cuatro variantes, que se diferencian en lo que recibe quien llega tarde:

const probar = (nombre: string, s: Subject<number>) => {
  s.next(1); s.next(2); s.next(3);

  const r: (number | string)[] = [];
  s.subscribe({ next: (v) => r.push(v), complete: () => r.push('fin') });

  s.next(4);
  s.complete();
  console.log(nombre.padEnd(16), r.join(' '));
};

probar('Subject', new Subject<number>());
probar('BehaviorSubject', new BehaviorSubject<number>(0));
probar('ReplaySubject(2)', new ReplaySubject<number>(2));
probar('AsyncSubject', new AsyncSubject<number>());

Subject          4 fin
BehaviorSubject  3 4 fin
ReplaySubject(2) 2 3 4 fin
AsyncSubject     4 fin

Figura 13. Las cuatro variantes frente a un suscriptor que llega después del 3.

Tipo	Al suscribirse recibes	Valor inicial	Uso típico
Subject	Nada, solo lo que venga	No	Eventos: “se ha guardado”
BehaviorSubject	El valor actual	Obligatorio	Estado con valor siempre
ReplaySubject(n)	Los n últimos	No	Historial, o caché sin valor inicial
AsyncSubject	El último, al completar	No	Rara vez hace falta

BehaviorSubject también permite leer el valor sin suscribirse, con .value o .getValue():

const tema$ = new BehaviorSubject<'claro' | 'oscuro'>('claro');
console.log(tema$.value);        // claro
tema$.next('oscuro');
console.log(tema$.getValue());   // oscuro

Un servicio con BehaviorSubject

Antes de las señales, este era el patrón estándar para compartir estado entre componentes, y lo encontrarás en muchísimo código. El servicio guarda el Subject en privado y publica solo la versión de lectura con asObservable(), para que nadie más pueda emitir:

// src/app/cesta.ts (el servicio)
import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable, map } from 'rxjs';
import { Producto } from './modelos';

export interface Linea {
  producto: Producto;
  unidades: number;
}

@Injectable({ providedIn: 'root' })
export class CestaServicio {
  // Privado: solo este servicio puede emitir
  private readonly lineasSubject = new BehaviorSubject<Linea[]>([]);

  // Público: los demás solo pueden suscribirse
  readonly lineas$: Observable<Linea[]> = this.lineasSubject.asObservable();

  readonly unidades$ = this.lineas$.pipe(
    map((lineas) => lineas.reduce((suma, l) => suma + l.unidades, 0)),
  );

  readonly total$ = this.lineas$.pipe(
    map((lineas) => lineas.reduce((suma, l) => suma + l.unidades * l.producto.precio, 0)),
  );

  anadir(producto: Producto) {
    const lineas = this.lineasSubject.value;
    const existe = lineas.some((l) => l.producto.productoId === producto.productoId);

    this.lineasSubject.next(
      existe
        ? lineas.map((l) =>
            l.producto.productoId === producto.productoId
              ? { ...l, unidades: l.unidades + 1 }
              : l,
          )
        : [...lineas, { producto, unidades: 1 }],
    );
  }
}

Dos componentes lo consumen con AsyncPipe. El resumen y la lista no saben nada el uno del otro:

// src/app/cesta.ts (los componentes; añade los imports de AsyncPipe, CurrencyPipe,
// HttpClient, Component e inject, y API y Pagina)
@Component({
  selector: 'app-resumen-cesta',
  imports: [AsyncPipe, CurrencyPipe],
  template: `
    <p class="resumen">
      Cesta: {{ cesta.unidades$ | async }} unidades,
      {{ cesta.total$ | async | currency: 'EUR' }}
    </p>
  `,
})
export class ResumenCesta {
  protected readonly cesta = inject(CestaServicio);
}

@Component({
  selector: 'app-tienda',
  imports: [AsyncPipe, CurrencyPipe, ResumenCesta],
  template: `
    <app-resumen-cesta />
    <ul>
      @for (p of productos$ | async; track p.productoId) {
        <li>
          {{ p.nombre }} ({{ p.precio | currency: 'EUR' }})
          <button type="button" (click)="cesta.anadir(p)">Añadir</button>
        </li>
      }
    </ul>
  `,
})
export class Tienda {
  protected readonly cesta = inject(CestaServicio);

  protected readonly productos$ = inject(HttpClient)
    .get<Pagina<Producto>>(`${API}/productos?tamano=3`)
    .pipe(map((pagina) => pagina.contenido));
}

Figura 14. Tras pulsar tres veces “Añadir” (dos en el primer producto y una en el segundo), el resumen se actualiza solo: 2 × 899 + 1499 = 3297.

💡 Consejo: En código nuevo, este estado sería una señal: signal<Linea[]>([]) más dos computed. Requiere menos código y no tiene suscripciones. Un BehaviorSubject sigue teniendo sentido si el estado alimenta una tubería con operadores de tiempo o aplanamiento. El módulo 19 trata este tema a fondo.

---

11. Frío y caliente

Esta distinción explica por qué a veces aparecen peticiones duplicadas sin motivo aparente.

• Un observable frío crea su productor en cada suscripción. Cada suscriptor tiene su propia ejecución. Es como una película bajo demanda: cada espectador la ve desde el principio.
• Un observable caliente tiene un productor que existe fuera de él y se reparte entre todos. Es como la televisión en directo: si llegas tarde, te pierdes lo anterior.

Figura 15. HttpClient, of, from, timer e interval son fríos; fromEvent y los Subject son calientes.

Una consecuencia directa es que suscribirte dos veces a una petición lanza dos peticiones:

let llamadas = 0;

const http$ = defer(() => {
  llamadas++;
  console.log(`Petición ${llamadas} al servidor`);
  return timer(50).pipe(map(() => ['Ventas', 'I+D']));
});

http$.subscribe();
http$.subscribe();

Petición 1 al servidor
Petición 2 al servidor

share: una ejecución para todos

share convierte un observable frío en caliente: el primer suscriptor arranca la ejecución y los demás se unen a ella.

const compartido$ = http$.pipe(share());

compartido$.subscribe((v) => console.log('A', v.length));
compartido$.subscribe((v) => console.log('B', v.length));

// Tras terminar la petición:
compartido$.subscribe((v) => console.log('C tardío', v.length));

Petición 1 al servidor
A 2
B 2
Petición 2 al servidor
C tardío 2

A y B comparten petición. El tardío, en cambio, lanza otra: cuando la fuente termina, share se reinicia y no guarda nada.

shareReplay: una ejecución y memoria

shareReplay añade un ReplaySubject por delante: quien llega tarde recibe los últimos valores sin repetir el trabajo. Es la herramienta habitual para datos que casi no cambian.

const conMemoria$ = http$.pipe(shareReplay({ bufferSize: 1, refCount: true }));

conMemoria$.subscribe((v) => console.log('A', v.length));
conMemoria$.subscribe((v) => console.log('B', v.length));

// Tras terminar la petición:
conMemoria$.subscribe((v) => console.log('C tardío', v.length));

Petición 1 al servidor
A 2
B 2
C tardío 2

⚠️ Importante: Usa siempre la forma de objeto con refCount: true. La forma corta shareReplay(1) equivale a refCount: false: deja la suscripción a la fuente activa aunque todos se den de baja. Con una petición no pasa nada, porque termina sola. Con un flujo infinito es una fuga.

Se ve con un reloj que avisa al arrancar y al pararse:

const reloj$ = interval(100).pipe(
  tap({
    subscribe: () => console.log('Reloj en marcha'),
    unsubscribe: () => console.log('Reloj parado'),
  }),
);

const r1 = reloj$.pipe(shareReplay({ bufferSize: 1, refCount: true })).subscribe();
// 150 ms después
r1.unsubscribe();

console.log('--- ahora con shareReplay(1)');
const r2 = reloj$.pipe(shareReplay(1)).subscribe();
// 150 ms después
r2.unsubscribe();
console.log('Baja hecha, pero el reloj sigue vivo');

Reloj en marcha
Reloj parado
--- ahora con shareReplay(1)
Reloj en marcha
Baja hecha, pero el reloj sigue vivo

connectable: controlar cuándo arranca

connectable crea un observable que no se pone en marcha con la primera suscripción, sino cuando llamas a connect(). Sirve si quieres que todos estén suscritos antes de empezar, incluso con una fuente síncrona:

const conectable = connectable(of('a', 'b').pipe(tap(() => console.log('produce'))), {
  connector: () => new Subject<string>(),
  resetOnDisconnect: false,
});

conectable.subscribe((v) => console.log('X', v));
conectable.subscribe((v) => console.log('Y', v));
console.log('Aún no ha pasado nada');
conectable.connect();

Aún no ha pasado nada
produce
X a
Y a
produce
X b
Y b

---

12. Fugas de suscripciones

Una fuga es una suscripción que sigue activa cuando ya nadie la necesita. El componente se destruye, pero la suscripción lo mantiene en memoria y sigue trabajando. Cada visita a la pantalla añade otra. La aplicación se vuelve lenta poco a poco, y aparecen efectos fantasma, como peticiones hechas por pantallas que ya no existen.

Este componente deja una fuga:

// src/app/fugas.ts
import { Component, DestroyRef, inject, signal } from '@angular/core';
import { Observable } from 'rxjs';

const vivasConFuga = signal(0);
const vivasLimpias = signal(0);

// Un reloj que lleva la cuenta de cuántas suscripciones siguen activas
function reloj(contador: typeof vivasConFuga) {
  return new Observable<number>((suscriptor) => {
    contador.update((n) => n + 1);
    const id = setInterval(() => suscriptor.next(Date.now()), 1000);
    return () => {
      clearInterval(id);
      contador.update((n) => n - 1);
    };
  });
}

@Component({ selector: 'app-reloj-con-fuga', template: `<p>Reloj con fuga: {{ hora() }}</p>` })
export class RelojConFuga {
  protected readonly hora = signal(0);

  constructor() {
    // Nadie llama a unsubscribe: la suscripción sobrevive al componente
    reloj(vivasConFuga).subscribe((t) => this.hora.set(t));
  }
}

@Component({ selector: 'app-reloj-limpio', template: `<p>Reloj limpio: {{ hora() }}</p>` })
export class RelojLimpio {
  protected readonly hora = signal(0);

  constructor() {
    const suscripcion = reloj(vivasLimpias).subscribe((t) => this.hora.set(t));
    inject(DestroyRef).onDestroy(() => suscripcion.unsubscribe());
  }
}

Un componente padre con un @if abre y cierra ambos relojes, y muestra los contadores. Al abrir y cerrar tres veces, el resultado no deja dudas:

Figura 16. Con los relojes ya cerrados, el componente con fuga mantiene tres temporizadores en marcha.

Quién necesita darse de baja

No todas las suscripciones requieren baja. La pregunta es: ¿este flujo termina solo, y antes de que deje de importarme?

Fuente	¿Termina sola?	¿Hay que darse de baja?
HttpClient	Sí, tras la respuesta	No por memoria. Sí si quieres cancelar al salir
of, from con datos	Sí	No
take(1), first()	Sí	Normalmente no
interval, timer periódico	Nunca	Sí
fromEvent	Nunca	Sí
Subject de un servicio	Cuando alguien lo complete	Sí
Router.events, valueChanges	Nunca, o con su dueño	Sí, si el dueño vive más que tú
\| async en la plantilla		No: lo hace el pipe

Las formas de darse de baja

1. AsyncPipe. Es la mejor si el valor acaba en la plantilla. No hay subscribe, así que no hay nada que olvidar.

2. unsubscribe con DestroyRef. Es lo que hace RelojLimpio. Sirve para casos sencillos. Con muchas suscripciones, agrupa con new Subscription() y add, como en la sección 2.

3. takeUntil con un notificador. Es el patrón clásico, y el que usa Buscador. Una vez que el notificador emite, takeUntil completa el flujo:

const destruido$ = new Subject<void>();

origen$.pipe(takeUntil(destruido$)).subscribe();
otro$.pipe(takeUntil(destruido$)).subscribe();

// Al destruir el componente
destruido$.next();
destruido$.complete();

4. takeUntilDestroyed. Es la versión moderna del patrón anterior. Se ocupa del notificador por ti, y lo verás en el módulo 15.

takeUntil va el último

La posición de takeUntil importa. Solo corta lo que tiene por encima en la tubería. Si lo pones antes de un operador de aplanamiento, la interna sobrevive:

function probar(nombre: string, construir: (o: Subject<number>, c: Subject<void>, i: () => Observable<number>) => Observable<number>) {
  const clics$ = new Subject<number>();
  const cierre$ = new Subject<void>();
  let internasVivas = 0;

  const interna = () =>
    interval(1000).pipe(
      tap({
        subscribe: () => internasVivas++,
        unsubscribe: () => internasVivas--,
      }),
    );

  construir(clics$, cierre$, interna).subscribe();
  clics$.next(1);
  cierre$.next();
  console.log(`${nombre}: ${internasVivas} interna viva`);
}

probar('takeUntil antes', (o, c, i) => o.pipe(takeUntil(c), switchMap(i)));
probar('takeUntil al final', (o, c, i) => o.pipe(switchMap(i), takeUntil(c)));

takeUntil antes: 1 interna viva
takeUntil al final: 0 interna viva

La explicación es que, al cortar arriba, switchMap solo ve que su fuente ha terminado. Pero un operador de aplanamiento no termina hasta que acaba su interna, y esta es infinita.

⚠️ Importante: Pon takeUntil como último operador de la tubería. Detrás de él solo deberían ir operadores que no abren suscripciones nuevas, como finalize.

---

13. Observables y promesas

A veces necesitas un resultado con await, por ejemplo al usar una biblioteca que trabaja con promesas. RxJS 7 ofrece dos funciones para ello:

import { EMPTY, EmptyError, firstValueFrom, interval, lastValueFrom, of } from 'rxjs';

console.log(await firstValueFrom(of('a', 'b', 'c')));   // a
console.log(await lastValueFrom(of('a', 'b', 'c')));    // c

try {
  await firstValueFrom(EMPTY);
} catch (e) {
  console.log(e instanceof EmptyError, (e as Error).message);
}

console.log(await firstValueFrom(EMPTY, { defaultValue: 'por defecto' }));
console.log('primer valor:', await firstValueFrom(interval(10)));

a
c
true no elements in sequence
por defecto
primer valor: 0

• firstValueFrom se resuelve con el primer valor y se da de baja. Es seguro incluso con flujos infinitos.
• lastValueFrom espera a que el flujo termine. Con un flujo infinito, la promesa no se resuelve nunca.
• Si el flujo termina sin emitir, ambas fallan con EmptyError, salvo que indiques defaultValue.

obs$.toPromise(), que verás en código antiguo, está obsoleto en RxJS 7. Para el sentido contrario, usa from(promesa), o mejor defer(() => promesa()), como viste en la sección 4.

---

14. Operadores propios

Un operador es una función de observable a observable. Hay dos maneras de crear uno, y la primera te bastará casi siempre.

Componer operadores existentes con pipe

Si repites la misma secuencia en varios sitios, dale un nombre. La función pipe de RxJS (no el método) combina operadores en uno solo:

import { distinctUntilChanged, filter, map, of, pipe, type OperatorFunction } from 'rxjs';

function soloTextoLimpio(minimo: number): OperatorFunction<string, string> {
  return pipe(
    map((t) => t.trim().toLowerCase()),
    filter((t) => t.length >= minimo),
    distinctUntilChanged(),
  );
}

of('  An', 'ANA', 'ana ', 'a', 'Luis').pipe(soloTextoLimpio(2)).subscribe(console.log);

an
ana
luis

Escribir uno desde cero

Si necesitas controlar la suscripción, recibe el origen y devuelve un new Observable. Debes reenviar los tres canales y dar de baja el origen en la limpieza. Este operador registra todo lo que pasa, y es muy útil para depurar:

function registrar<T>(etiqueta: string): MonoTypeOperatorFunction<T> {
  return (origen) =>
    new Observable<T>((destino) => {
      console.log(`[${etiqueta}] suscripción`);
      const interior = origen.subscribe({
        next: (v) => { console.log(`[${etiqueta}] valor`, v); destino.next(v); },
        error: (e) => { console.log(`[${etiqueta}] error`, e); destino.error(e); },
        complete: () => { console.log(`[${etiqueta}] fin`); destino.complete(); },
      });
      return () => {
        console.log(`[${etiqueta}] baja`);
        interior.unsubscribe();
      };
    });
}

of(1, 2).pipe(registrar('prueba')).subscribe();

[prueba] suscripción
[prueba] valor 1
[prueba] valor 2
[prueba] fin
[prueba] baja

MonoTypeOperatorFunction<T> es un OperatorFunction<T, T>: la salida tiene el mismo tipo que la entrada.

---

15. Probar flujos con canicas

Probar código con tiempo usando esperas reales es lento y poco fiable. TestScheduler ejecuta los flujos con un reloj virtual: 300 ms de debounceTime pasan al instante y siempre igual. Es lo que has visto funcionar en los experimentos de este módulo.

Dentro de run, escribes los flujos con cadenas de canicas. Las herramientas de pruebas se explican en el módulo 26. Este fichero funciona con ng test:

// src/app/canicas.spec.ts
import { debounceTime, filter, map, retry, switchMap } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';

function crearPlanificador() {
  return new TestScheduler((real, esperado) => expect(real).toEqual(esperado));
}

describe('canicas', () => {
  it('map y filter', () => {
    crearPlanificador().run(({ cold, expectObservable }) => {
      const origen = cold('-a-b-c-d-|', { a: 1, b: 2, c: 3, d: 4 });
      const resultado = origen.pipe(
        filter((n) => n % 2 === 0),
        map((n) => n * 10),
      );
      expectObservable(resultado).toBe('---x---y-|', { x: 20, y: 40 });
    });
  });

  it('debounceTime con tiempo virtual', () => {
    crearPlanificador().run(({ cold, expectObservable }) => {
      const teclas = cold('a 100ms b 100ms c 999ms |');
      const resultado = teclas.pipe(debounceTime(300));
      // c llega en 202 y se emite 300 ms después; el fin llega en 1202
      expectObservable(resultado).toBe('502ms c 699ms |');
    });
  });

  it('switchMap cancela la interna', () => {
    crearPlanificador().run(({ cold, hot, expectObservable, expectSubscriptions }) => {
      const origen = hot('  --a-b--------|');
      const interna = cold('  xx-y|');
      const resultado = origen.pipe(switchMap(() => interna));

      expectObservable(resultado).toBe('--xxxx-y-----|');
      expectSubscriptions(interna.subscriptions).toBe([
        '--^-!',
        '----^---!',
      ]);
    });
  });

  it('retry vuelve a suscribirse', () => {
    crearPlanificador().run(({ cold, expectObservable, expectSubscriptions }) => {
      const fuente = cold('--a--#');
      expectObservable(fuente.pipe(retry(1))).toBe('--a----a--#');
      expectSubscriptions(fuente.subscriptions).toBe(['^----!', '-----^----!']);
    });
  });
});

Símbolo	Significado
-	Pasa 1 ms de tiempo virtual
a	Emite un valor; con un mapa de valores, el que corresponda
\|	Fin del flujo
#	Error
( )	Agrupa emisiones en el mismo instante: (ab\|)
100ms	Salto de tiempo, siempre rodeado de espacios
^	Momento de la suscripción (en hot y en suscripciones)
!	Momento de la baja

Tres detalles que suelen confundir:

• Una letra también ocupa tiempo. En a 100ms b, a está en 0 y ocupa 1 ms, así que b llega en 101. Por eso la prueba de debounceTime tiene esos números.
• cold y hot son los mismos conceptos de la sección 11. cold empieza a contar al suscribirse y hot tiene su propia línea temporal.
• expectSubscriptions comprueba cuándo se suscribe y se da de baja el operador. Es la forma de demostrar que switchMap cancela: la primera interna se corta en el instante 4, sin llegar a emitir y.

---

16. Errores frecuentes

Síntoma	Causa probable	Solución
No pasa nada, ni siquiera la petición	Nadie se suscribe	Añade subscribe o usa async
La petición se lanza dos veces	Dos suscripciones a un flujo frío	Una sola async con @if (x$ \| async; as x), o shareReplay
El buscador deja de responder tras un error	catchError fuera del aplanador	Muévelo dentro
Resultados antiguos pisan a los nuevos	mergeMap en una búsqueda	Usa switchMap
Un doble clic crea dos registros	mergeMap o concatMap al enviar	Usa exhaustMap
Guardados perdidos	switchMap en una escritura	Usa concatMap
combineLatest no emite	Una entrada aún no ha emitido	startWith o BehaviorSubject
forkJoin no emite	Un flujo no termina, o está vacío	Comprueba las entradas; take(1)
reduce o lastValueFrom no acaban	El flujo es infinito	Usa scan o firstValueFrom
La vista no se actualiza tras subscribe	Se asigna a una propiedad normal	Asigna a una señal o usa async
La memoria crece al navegar	Suscripciones sin baja	Sección 12
El flujo sigue tras takeUntil	takeUntil antes del aplanador	Colócalo el último
Observable<Observable<T>>	map en lugar de un aplanador	Cambia a switchMap

💡 Consejo: Para depurar una tubería, añade tap entre operadores o usa el operador registrar de la sección 14. Te dirá si el flujo llega hasta allí, si termina y cuándo se da de baja.

---

17. Ejercicios

Ejercicio 1: Cuenta atrás

Crea un flujo que emita 5, 4, 3, 2, 1 con un segundo de separación, empezando en el acto, y que termine con el texto ¡Despegue!. Usa timer, map, takeWhile y endWith.

Ver solución

import { endWith, map, takeWhile, timer } from 'rxjs';

timer(0, 1000).pipe(
  map((i) => 5 - i),
  takeWhile((n) => n > 0),
  map(String),
  endWith('¡Despegue!'),
).subscribe(console.log);

Salida, con el instante de cada línea medido en reloj virtual:

0 5
1000 4
2000 3
3000 2
4000 1
5000 ¡Despegue!

timer(0, 1000) emite 0, 1, 2… empezando ahora. takeWhile completa el flujo al llegar el 0 sin dejarlo pasar, y endWith añade un valor justo al completarse. El map(String) hace que todos los valores sean del mismo tipo.

Ejercicio 2: Resumen de pedidos

Dado un flujo de pedidos con id e importe, obtén en cada emisión el número de pedidos, el total acumulado y el mayor importe visto hasta ese momento.

Ver solución

import { of, scan } from 'rxjs';

interface Pedido { id: number; importe: number; }

const pedidos$ = of<Pedido[]>(
  { id: 5001, importe: 120 },
  { id: 5002, importe: 80.5 },
  { id: 5003, importe: 299.5 },
);

pedidos$.pipe(
  scan(
    (acumulado, pedido) => ({
      pedidos: acumulado.pedidos + 1,
      total: acumulado.total + pedido.importe,
      mayor: Math.max(acumulado.mayor, pedido.importe),
    }),
    { pedidos: 0, total: 0, mayor: 0 },
  ),
).subscribe((r) => console.log(r));

{ pedidos: 1, total: 120, mayor: 120 }
{ pedidos: 2, total: 200.5, mayor: 120 }
{ pedidos: 3, total: 500, mayor: 299.5 }

scan encaja porque quieres cada estado intermedio. Con reduce, solo verías la última línea.

Ejercicio 3: Elige el aplanador

Indica qué operador usarías en cada caso y por qué:

1. Autocompletado de nombres de clientes.
2. Un botón “Confirmar pedido”, en el que el usuario puede pulsar varias veces.
3. Autoguardado de un borrador cada vez que cambia, con versiones que deben llegar en orden.
4. Descargar en paralelo las fichas de cuarenta empleados, sin saturar el servidor.
5. Pulsar “Siguiente” en un visor de pedidos, que muestra el pedido actual.

Ver solución

Caso	Operador	Motivo
1	switchMap	Lectura; solo importa lo último, y evita la carrera
2	exhaustMap	Ignora los clics mientras el pedido está en marcha
3	concatMap	Un guardado no debe adelantar al anterior
4	mergeMap(f, 4)	Tareas independientes, con concurrencia limitada
5	switchMap	Si el usuario sigue avanzando, el pedido anterior ya no interesa

En el caso 2, concatMap crearía un pedido por cada clic. Y con switchMap, cancelar un POST deja el resultado en el aire: el servidor puede haberlo procesado igualmente.

Ejercicio 4: Panel tolerante a fallos

Crea un componente Panel que cargue en paralelo las estadísticas, los departamentos y los proyectos. Si fallan los proyectos, el panel debe mostrar el resto y un aviso. Para provocar el fallo, pide los proyectos a una ruta que no exista.

Ver solución

// src/app/panel.ts
import { AsyncPipe, CurrencyPipe } from '@angular/common';
import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { catchError, forkJoin, map, of } from 'rxjs';
import { API, Departamento, Estadisticas, Pagina, Proyecto } from './modelos';

@Component({
  selector: 'app-panel',
  imports: [AsyncPipe, CurrencyPipe],
  template: `
    @if (panel$ | async; as panel) {
      <p>
        {{ panel.estadisticas.empleados }} empleados,
        {{ panel.departamentos.length }} departamentos,
        {{ panel.estadisticas.facturacion | currency: 'EUR' }} facturados
      </p>
      @if (panel.proyectos; as proyectos) {
        <p>{{ proyectos.length }} proyectos activos</p>
      } @else {
        <p class="error">Los proyectos no están disponibles ahora mismo</p>
      }
    } @else {
      <p class="aviso">Cargando el panel...</p>
    }
  `,
})
export class Panel {
  private readonly http = inject(HttpClient);

  protected readonly panel$ = forkJoin({
    estadisticas: this.http.get<Estadisticas>(`${API}/estadisticas`),
    departamentos: this.http
      .get<Pagina<Departamento>>(`${API}/departamentos`)
      .pipe(map((pagina) => pagina.contenido)),
    proyectos: this.http
      .get<Pagina<Proyecto>>(`${API}/ruta-que-no-existe`)
      .pipe(
        map((pagina) => pagina.contenido),
        catchError(() => of(null)),
      ),
  });
}

Figura 17. Una ruta falla y el resto del panel se muestra con normalidad.

La clave es el catchError dentro de la entrada de forkJoin. Sin él, el 404 haría fallar el forkJoin completo. null es un buen sustituto porque distingue “no ha cargado” de “no hay proyectos”. En el tipo de panel$, proyectos queda como Proyecto[] | null, así que TypeScript obliga a tratar los dos casos. Cambia la ruta a ${API}/proyectos para ver el caso normal.

Ejercicio 5: Reintentos solo si compensa

Escribe un operador reintentarSiServidor(veces, base) que reintente solo los errores 5xx, con espera exponencial a partir de base. Un error 4xx debe propagarse sin más intentos.

Ver solución

import { defer, of, retry, throwError, timer, type MonoTypeOperatorFunction } from 'rxjs';

interface ErrorHttp { status: number; }

function reintentarSiServidor<T>(veces: number, base: number): MonoTypeOperatorFunction<T> {
  return retry({
    count: veces,
    delay: (error: ErrorHttp, intento) =>
      error.status >= 500
        ? timer(base * 2 ** (intento - 1))
        : throwError(() => error),
  });
}

delay devuelve un observable que decide: si emite, se reintenta; si da error, ese error se propaga. Con reintentarSiServidor(3, 200) y tres respuestas simuladas, el reloj virtual da:

0 ms: intento 1 -> 503
200 ms: intento 2 -> 503
600 ms: intento 3 -> 200
600 ms: datos

0 ms: intento 1 -> 503
200 ms: intento 2 -> 404
200 ms: me rindo con 404

0 ms: intento 1 -> 500
200 ms: intento 2 -> 500
600 ms: intento 3 -> 500
1400 ms: intento 4 -> 500
1400 ms: me rindo con 500

HttpErrorResponse tiene la propiedad status, así que el operador se aplica tal cual a una petición de HttpClient.

Ejercicio 6: Caché de departamentos

Varias pantallas piden la lista de departamentos. Crea un servicio que haga una sola petición aunque haya varios suscriptores, también si llegan más tarde. Comprueba que, si la petición falla, un intento posterior vuelve a llamar al servidor.

Ver solución

// src/app/servicios.ts
import { HttpClient } from '@angular/common/http';
import { Injectable, inject } from '@angular/core';
import { Observable, map, shareReplay } from 'rxjs';
import { API, Departamento, Pagina } from './modelos';

@Injectable({ providedIn: 'root' })
export class DepartamentosServicio {
  private readonly http = inject(HttpClient);

  readonly departamentos$: Observable<Departamento[]> = this.http
    .get<Pagina<Departamento>>(`${API}/departamentos`)
    .pipe(
      map((pagina) => pagina.contenido),
      shareReplay({ bufferSize: 1, refCount: true }),
    );
}

Dos cosas importantes:

• El observable es una propiedad, creada una sola vez. Si fuera un método departamentos() con shareReplay dentro, cada llamada crearía una caché nueva y no compartiría nada.
• Aunque refCount sea true, los suscriptores tardíos reciben el valor sin otra petición, porque la fuente ya ha terminado, como viste en la sección 11.

Esta prueba, que pasa con ng test, comprueba ambos comportamientos con HttpTestingController. El error no queda guardado, y por eso se puede reintentar:

it('caché: varias suscripciones, una petición', () => {
  const s = TestBed.inject(DepartamentosServicio);
  const r: number[] = [];
  s.departamentos$.subscribe((l) => r.push(l.length));
  s.departamentos$.subscribe((l) => r.push(l.length));
  http.expectOne(`${API}/departamentos`).flush({ contenido: [{ deptId: 10 }, { deptId: 20 }] });

  s.departamentos$.subscribe((l) => r.push(l.length));
  http.expectNone(`${API}/departamentos`);
  expect(r).toEqual([2, 2, 2]);
});

it('caché: un error no se queda guardado', () => {
  const s = TestBed.inject(DepartamentosServicio);
  let error: HttpErrorResponse | null = null;
  s.departamentos$.subscribe({ error: (e) => (error = e) });
  http.expectOne(`${API}/departamentos`).flush('fallo', { status: 500, statusText: 'Error' });
  expect(error).not.toBeNull();

  let n = 0;
  s.departamentos$.subscribe((l) => (n = l.length));
  http.expectOne(`${API}/departamentos`).flush({ contenido: [{ deptId: 10 }] });
  expect(n).toBe(1);
});

Ejercicio 7: Suscripciones anidadas

Reescribe el ejemplo con subscribe anidado de la sección 6 como un método ficha(id) que devuelva un Observable<FichaCompleta> con el nombre del empleado y la localidad de su departamento. Comprueba que darse de baja cancela la petición.

Ver solución

export interface FichaCompleta {
  nombre: string;
  localidad: string;
}

@Injectable({ providedIn: 'root' })
export class FichasServicio {
  private readonly http = inject(HttpClient);

  ficha(id: number): Observable<FichaCompleta> {
    return this.http.get<Empleado>(`${API}/empleados/${id}`).pipe(
      switchMap((empleado) =>
        this.http.get<Departamento>(`${API}/departamentos/${empleado.deptId}`).pipe(
          map((departamento) => ({
            nombre: `${empleado.nombre} ${empleado.apellidos}`,
            localidad: departamento.localidad,
          })),
        ),
      ),
    );
  }
}

El map va dentro del switchMap porque ahí sigue accesible empleado, gracias al cierre léxico. Ahora quien llama tiene una sola suscripción, un solo punto de error y una baja que alcanza a todo. Lo comprueba esta prueba:

it('cancelar marca la petición como cancelada', () => {
  const s = TestBed.inject(FichasServicio);
  const sus = s.ficha(100).subscribe();
  const req = http.expectOne(`${API}/empleados/100`);
  sus.unsubscribe();
  expect(req.cancelled).toBe(true);
});

Con una sola respuesta por petición, switchMap, concatMap y mergeMap dan aquí el mismo resultado. Se elige switchMap por costumbre: es el más seguro para lecturas.

Ejercicio 8: Probar la tubería del buscador

Escribe una prueba de canicas para una tubería con debounceTime(300), distinctUntilChanged, switchMap y catchError dentro, y takeUntil. Las teclas son: a en 0, an en 2, de nuevo an en 503 y ana en 1004. La API tarda 2 ms; con ana da error. El cierre llega en 2 s.

Ver solución

it('ejercicio: tubería del buscador', () => {
  crearPlanificador().run(({ hot, cold, expectObservable }) => {
    const teclas = hot('a-b 500ms b 500ms c', { a: 'a', b: 'an', c: 'ana' });
    const api = (texto: string) =>
      texto === 'ana'
        ? cold('--#', {}, new Error('500'))
        : cold('--(r|)', { r: `resultado ${texto}` });
    const fin = hot('2s x');

    const resultado = teclas.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap((t) => api(t).pipe(catchError(() => of('error')))),
      takeUntil(fin),
    );

    expectObservable(resultado).toBe('304ms r 1001ms e 693ms |', {
      r: 'resultado an',
      e: 'error',
    });
  });
});

El razonamiento, instante a instante:

• an llega en 2. Tras 300 ms de calma sale en 302, y la API responde en 304.
• En 503 se repite an. Sale de debounceTime en 803, pero distinctUntilChanged lo descarta y no hay petición.
• ana llega en 1004 y sale en 1304. La API falla en 1306, y catchError lo convierte en 'error'.
• El cierre en 2000 completa el flujo. Ese fin en 2000 es la prueba de que el error no mató la tubería.

Cuenta las posiciones en la cadena esperada: r está en 304, ocupa 1 ms, y 1001ms lleva a 1306; e ocupa 1 ms y 693ms lleva a 2000.

Ejercicio 9 (propuesto): Inicio de sesión sin doble envío

Crea un formulario con email y contraseña que envíe POST /api/auth/login. Usa un Subject para los envíos y exhaustMap para ignorar los clics repetidos. Un error de credenciales debe mostrar un mensaje sin impedir un segundo intento. Usuario de prueba: ana.garcia@empresa.es con contraseña secreto. Pista: ¿dónde va el catchError?

Ejercicio 10 (propuesto): Autoguardado

Crea un área de texto que guarde su contenido en el servidor, con debounceTime(1000), distinctUntilChanged y concatMap. Simula el guardado con timer y muestra “Guardando…” y “Guardado”. Después sustituye concatMap por switchMap, y explica qué podría perderse con una red lenta.

Ejercicio 11 (propuesto): Sondeo de estadísticas

Refresca GET /api/estadisticas cada 10 segundos con timer(0, 10000) y switchMap, y añade un botón de pausa con BehaviorSubject<boolean>. Pista: combina el estado de pausa con switchMap y EMPTY. Asegúrate de que el sondeo se detiene al destruir el componente, y arranca la API con RETARDO=2000 para ver qué ocurre con las peticiones lentas.

---

Resumen

Concepto	API	Idea clave
Flujo	Observable	Una receta perezosa de valores en el tiempo
Consumidor	Observer	next, error y complete
Ejecución	Subscription	unsubscribe() corta y limpia
Contrato	next* (error \| complete)	Tras un error o un fin no llega nada más
Creación	of, from, timer, interval, fromEvent, defer	Casi nunca hace falta new Observable
Encadenar	pipe	Cada operador crea un observable nuevo
Básicos	map, filter, tap, scan, take, distinctUntilChanged	Cada valor recorre toda la tubería
Aplanar: última	switchMap	Cancela la anterior; para lecturas
Aplanar: cola	concatMap	Conserva el orden; para guardados
Aplanar: paralelo	mergeMap	Sin orden; admite un límite de concurrencia
Aplanar: ignora	exhaustMap	Evita el doble envío
Tiempo	debounceTime, throttleTime, auditTime, sampleTime	Cuándo se deja pasar un valor
Combinar	combineLatest, withLatestFrom, forkJoin, zip, merge, concat, race	Qué dispara la salida
Errores	catchError, retry, timeout, finalize	Captura dentro del aplanador
Multidifusión	Subject, BehaviorSubject, ReplaySubject, AsyncSubject	Observable y observador a la vez
Compartir	share, shareReplay, connectable	Con refCount: true en shareReplay
Bajas	async, takeUntil, DestroyRef, Subscription.add	takeUntil va el último
Promesas	firstValueFrom, lastValueFrom	toPromise está obsoleto
Propios	pipe(), OperatorFunction	Compón antes de escribir desde cero
Pruebas	TestScheduler, cold, hot, expectObservable	El tiempo virtual da pruebas exactas e instantáneas

---

Navegación
← Módulo anterior: Resources	Siguiente módulo: RxJS y señales →