← Volver al listado de tecnologías ← Índice de Angular: Buenas Prácticas

RxJS con criterio

Por: Artiko
angularbuenas-practicasrxjssignalsrxjs-interopmemory-leaksreactividad

RxJS con criterio

Con los signals estables, mucha gente sacó la conclusión equivocada: “RxJS ya no hace falta”. Es un error de categoría. Signals resuelven estado: un valor que existe ahora, que se lee de forma síncrona y que propaga cambios a quien lo derive. RxJS resuelve eventos en el tiempo: cosas que ocurren, con orden, con latencia, con posibilidad de fallar, de cancelarse y de coordinarse entre sí. No compiten; cubren dos ejes distintos.

Este capítulo va sobre el criterio: cuándo todavía necesitás observables, qué operadores usar sin meterte un bug de concurrencia, cómo dejar de tener memory leaks por suscripciones colgadas, y cómo cruzar el puente entre ambos mundos sin construir un Frankenstein. El capítulo anterior, Signals y estado reactivo, cubre el otro lado; este asume que ya lo tenés claro.

El eje temporal: qué solo RxJS hace bien

Un signal siempre tiene un valor y siempre te da el último. No tiene noción de “cada vez que pasó X”, ni de “esperá 300 ms”, ni de “reintentá tres veces”. En cuanto tu problema tiene una dimensión temporal o de secuencia de eventos, RxJS es la herramienta correcta:

La heurística: si te importa el valor actual, es un signal; si te importa la secuencia de lo que pasó y cuándo, es un observable. Un typeahead de búsqueda es el ejemplo canónico donde ambos se tocan, y lo usamos de hilo conductor.

flowchart TD
  Q{"¿Qué modelás?"}
  Q -->|"Un valor que existe ahora<br/>y se lee síncrono"| S["signal / computed"]
  Q -->|"Eventos en el tiempo,<br/>con orden y latencia"| R["Observable (RxJS)"]
  R --> R1["debounce / throttle / retry"]
  R --> R2["WebSocket / SSE / eventos DOM"]
  R --> R3["cancelación con switchMap"]
  R --> Bridge["toSignal() en el borde"]
  Bridge --> V["El template consume un signal"]
  S --> V

Fijate el patrón que emerge: RxJS vive en el borde (la fuente de eventos y su coordinación) y el resultado se convierte a signal para que el template lo consuma. Casi nunca querés observables crudos en el HTML.

Operadores de aplanamiento: switchMap vs mergeMap vs concatMap vs exhaustMap

Cuando un observable emite valores y cada valor dispara otro observable (típicamente una request HTTP), tenés un observable de observables. Aplanarlo es el 80% de RxJS en una app Angular, y elegir mal el operador es el bug de concurrencia más común del ecosistema. Los cuatro se diferencian en qué hacen cuando llega un valor nuevo mientras el interno anterior sigue vivo.

OperadorAl llegar un valor nuevoConcurrenciaCaso de uso
switchMapCancela el interno anterior y arranca el nuevo1 (el último)Búsqueda / typeahead, cargar detalle del ítem seleccionado
mergeMapArranca el nuevo sin cancelar nadaN (todos a la vez)Operaciones independientes en paralelo sin orden
concatMapEncola y espera a que termine el anterior1 (en orden)Escrituras que deben respetar el orden (guardar una cola)
exhaustMapIgnora el nuevo hasta que termine el actual1 (el primero)Botón de login/submit: bloquear el doble click

El bug clásico: race condition con mergeMap en búsquedas

Este es el error que aparece en revisiones de código una y otra vez. Una búsqueda con mergeMap:

// ANTES (antipatrón): race condition
searchControl.valueChanges.pipe(
  debounceTime(300),
  mergeMap(term => this.api.search(term)) // ← dispara N requests en paralelo
).subscribe(results => this.results.set(results));

El problema es de complejidad temporal, no de sintaxis: mergeMap no cancela. Si escribís "ang" y después "angular", salen dos requests. Si la de "ang" (más lenta, quizá el servidor la atendió después) responde última, sobreescribe los resultados de "angular". El usuario ve resultados de "ang" mientras el input dice "angular". El bug es intermitente porque depende del orden de llegada de la red: imposible de reproducir en local, un infierno en producción.

// DESPUÉS (mejor): switchMap cancela la request obsoleta
searchControl.valueChanges.pipe(
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(term => this.api.search(term)) // ← cancela la anterior en vuelo
).subscribe(results => this.results.set(results));

switchMap desuscribe del observable interno anterior en cuanto llega un término nuevo. En HTTP, eso aborta la request obsoleta (Angular emite el abort del fetch/XHR). Solo sobrevive la última: el resultado siempre corresponde al término actual. La regla mental: para búsquedas y navegación a “lo último seleccionado”, switchMap casi siempre.

sequenceDiagram
  participant U as Usuario
  participant S as switchMap
  participant API as API
  U->>S: "ang" (debounce 300ms)
  S->>API: GET /search?q=ang
  U->>S: "angular"
  S-->>API: cancel GET q=ang
  S->>API: GET /search?q=angular
  API-->>S: resultados de "angular"
  S-->>U: render (coherente con el input)

El costado de rendimiento importa: con switchMap hacés como máximo una request “viva” por ráfaga de tecleo. Con mergeMap hacés una por cada término intermedio que sobreviva al debounce: más carga en el servidor, más ancho de banda y el race condition de regalo.

Cuándo cada uno de los otros tres

Coordinación temporal: los operadores que valen oro

Estos son los que justifican traer RxJS aunque el resto de tu app sea signals:

input.valueChanges.pipe(
  debounceTime(300),          // espera 300ms de silencio antes de emitir
  distinctUntilChanged(),     // no reacciona si el valor no cambió
  filter(term => term.length >= 2),
  switchMap(term => this.api.search(term).pipe(
    retry({ count: 2, delay: 500 }),   // reintenta ante fallo transitorio
    catchError(() => of([]))           // degradá con gracia; no rompas el stream
  ))
);

Un detalle de ubicación que causa bugs: el catchError/retry va dentro del switchMap, envolviendo la request interna, no en el stream externo. Si lo ponés afuera, un solo error igual mata el stream de valueChanges y el input deja de buscar.

shareReplay y su fuga de memoria

shareReplay multicastea un observable (una sola ejecución del source compartida entre suscriptores) y replaya los últimos N valores a quien llegue tarde. Es ideal para cachear una request que varios componentes consumen. El problema es la configuración por defecto.

// ANTES (antipatrón): fuga de memoria
readonly config$ = this.http.get<Config>('/api/config').pipe(
  shareReplay(1) // refCount por defecto es false
);

Con refCount: false, la suscripción al source nunca se desconecta aunque todos los suscriptores se vayan. Para una request HTTP que completa, el efecto no es que el fetch quede vivo (ya completó), sino que el operador retiene para siempre el valor buferizado y la cadena de suscripción interna: no se libera aunque nadie lo use. En un observable de larga vida (un WebSocket, un interval) es peor: el source sigue ejecutándose indefinidamente porque el refCount no baja a cero nunca.

// DESPUÉS (mejor): refCount cierra el source cuando nadie escucha
readonly config$ = this.http.get<Config>('/api/config').pipe(
  shareReplay({ bufferSize: 1, refCount: true })
);

Con refCount: true, el operador cuenta suscriptores: cuando el último se va, desuscribe del source; si llega uno nuevo, re-ejecuta. Es lo que querés casi siempre. La excepción es un caché deliberadamente eterno (config global que se pide una vez y vive toda la sesión): ahí refCount: false es intencional, pero documentalo. Regla: shareReplay sin objeto de config explícito es una señal de alarma en revisión de código.

Memory leaks: el problema y su cura moderna

El leak clásico de Angular es la suscripción manual que nadie cancela:

// ANTES (antipatrón obsoleto): subscribe manual sin cleanup
export class PricesComponent implements OnInit, OnDestroy {
  private sub?: Subscription;

  ngOnInit() {
    this.sub = this.socket.prices$.subscribe(p => this.prices.set(p));
  }

  ngOnDestroy() {
    this.sub?.unsubscribe(); // fácil de olvidar; se multiplica por cada subscribe
  }
}

Si te olvidás del unsubscribe (o agregás un segundo subscribe y no lo sumás al cleanup), el callback sigue vivo tras destruir el componente: retiene el componente entero en memoria y sigue ejecutando lógica sobre un componente muerto. En un WebSocket que emite cada segundo, cada navegación deja un componente zombie escuchando. El leak crece con el uso.

El patrón intermedio takeUntil(this.destroy$) con un Subject manual también quedó obsoleto:

// ANTES (patrón de transición, hoy innecesario): Subject + takeUntil manual
private destroy$ = new Subject<void>();
ngOnInit() {
  this.socket.prices$.pipe(takeUntil(this.destroy$)).subscribe(/* ... */);
}
ngOnDestroy() {
  this.destroy$.next();
  this.destroy$.complete();
}

Funciona, pero es boilerplate: un Subject, un ngOnDestroy, y el riesgo de poner el takeUntil en el lugar equivocado del pipe. Angular moderno tiene tres formas mejores, en orden de preferencia.

1. toSignal(): el borde de RxJS al template

Si el observable alimenta la vista, convertilo a signal en el borde. toSignal se desuscribe solo cuando el contexto de inyección se destruye:

// DESPUÉS (mejor): cero cleanup manual
export class PricesComponent {
  private socket = inject(PriceSocket);
  readonly prices = toSignal(this.socket.prices$, { initialValue: [] });
  // el template usa {{ prices() }} — no hay subscribe, no hay leak
}

Opciones de toSignal que importan:

Restricción clave: toSignal requiere contexto de inyección (campo de clase o dentro de un inject()/constructor), salvo que le pases injector o manualCleanup.

2. async pipe: si el observable solo se consume en el template

El async pipe se suscribe y desuscribe con el ciclo de vida del template. En apps con signals suele ganar toSignal (podés derivar con computed, leerlo en TS), pero para un observable de un solo uso en la vista sigue siendo válido:

@if (data$ | async; as data) {
  <app-detalle [data]="data" />
}

Nunca uses varios | async del mismo observable en un template sin shareReplay: cada async crea una suscripción nueva y, sin multicast, N ejecuciones del source (N requests HTTP). El patrón *ngIf="... as x" (hoy @if (... ; as x)) resuelve una vez y reutiliza.

3. takeUntilDestroyed(): para side effects, no para la vista

Cuando el observable no alimenta la vista sino que dispara un efecto (loguear, sincronizar, navegar), y sí o sí necesitás un subscribe, atalo al ciclo de vida con takeUntilDestroyed():

export class TrackingComponent {
  constructor() {
    // en contexto de inyección: toma el DestroyRef automáticamente
    this.analytics.events$.pipe(
      takeUntilDestroyed()
    ).subscribe(e => this.reportar(e));
  }

  private startFuera() {
    // fuera de contexto de inyección: pasale el DestroyRef explícito
    const destroyRef = inject(DestroyRef);
    this.socket.prices$.pipe(
      takeUntilDestroyed(destroyRef)
    ).subscribe(/* ... */);
  }
}

takeUntilDestroyed() reemplaza todo el patrón Subject + ngOnDestroy: usa el DestroyRef del contexto actual. Sin argumento, debe llamarse en contexto de inyección (constructor o campo). Fuera de él, pasale un DestroyRef obtenido con inject(DestroyRef).

flowchart TD
  O["Tengo un Observable"] --> Q1{"¿Alimenta la vista?"}
  Q1 -->|Sí| Q2{"¿Necesito derivar/leer en TS?"}
  Q2 -->|Sí| TS["toSignal()"]
  Q2 -->|"No, solo template"| AP["async pipe"]
  Q1 -->|"No, es un side effect"| TUD["subscribe + takeUntilDestroyed()"]
  style TS fill:#2d6a4f,color:#fff
  style AP fill:#40916c,color:#fff
  style TUD fill:#52796f,color:#fff

Interoperabilidad: cuándo cruzar el puente

Angular ofrece toSignal y toObservable (paquete @angular/core/rxjs-interop) para cruzar entre mundos. La regla es cruzar en el borde, una vez, no ida y vuelta.

toObservable(signal): cuando necesitás el eje temporal sobre un signal

toObservable convierte un signal en observable usando un effect interno. Emite de forma asíncrona tras estabilizar el signal: si actualizás el signal tres veces en un tick, emite una sola vez con el valor final (coalescing). Es el puente para aplicarle operadores temporales a un valor de estado:

// El query es estado (signal); la búsqueda es temporal (RxJS)
readonly query = signal('');

readonly results = toSignal(
  toObservable(this.query).pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap(q => q ? this.api.search(q) : of([]))
  ),
  { initialValue: [] }
);

Acá el flujo es signal → observable → (coordinación temporal) → signal. El template escribe query con [(ngModel)] o un handler, y lee results(). RxJS aparece solo en el medio, donde aporta el debounce y la cancelación con switchMap. Esto es exactamente lo que querés.

Cuándo NO cruzar

// ANTES (antipatrón): RxJS para lo que computed resuelve
readonly total$ = toObservable(this.items).pipe(
  map(items => items.reduce((s, i) => s + i.precio, 0))
);
// DESPUÉS (mejor): computed puro
readonly total = computed(() => this.items().reduce((s, i) => s + i.precio, 0));

rxResource: HTTP asíncrono con cancelación integrada

Cuando la fuente asíncrona es una request que depende de params reactivos, rxResource combina lo mejor de ambos: params como signals, stream como factory de observable, y estados value/error/isLoading como signals, con cancelación automática al cambiar los params (equivalente a switchMap sin escribirlo):

readonly userId = signal(1);
readonly user = rxResource({
  params: () => ({ id: this.userId() }),
  stream: ({ params }) => this.http.get<User>(`/api/users/${params.id}`),
});
// user.value() / user.isLoading() / user.error() son signals

Para la mayoría de las cargas de datos ligadas a estado, rxResource (o httpResource, cubierto en HTTP, datos y complejidad) evita que escribas el pipeline de aplanamiento a mano.

Antipatrones a erradicar

subscribe dentro de subscribe

El síntoma más claro de no entender el aplanamiento:

// ANTES (antipatrón): callback hell + sin cancelación + doble leak
this.route.params.subscribe(params => {
  this.api.getUser(params.id).subscribe(user => {
    this.api.getOrders(user.id).subscribe(orders => {
      this.orders.set(orders);
    });
  });
});

Cada subscribe anidado es una suscripción independiente que nadie cancela, sin manejo de errores, y con race conditions si params cambia rápido. Aplaná con el operador correcto:

// DESPUÉS (mejor): un solo pipe, cancelación con switchMap, un solo cleanup
readonly orders = toSignal(
  this.route.params.pipe(
    switchMap(params => this.api.getUser(params.id)),
    switchMap(user => this.api.getOrders(user.id)),
    catchError(() => of([]))
  ),
  { initialValue: [] }
);

Lógica de negocio dentro del subscribe

El subscribe debería ser el punto final que consume el resultado, no donde vive la transformación. Toda la lógica (filtrar, mapear, combinar) va en operadores del pipe, que son testeables y componibles:

// ANTES: lógica escondida en el subscribe, imposible de testear en aislamiento
source$.subscribe(data => {
  const activos = data.filter(x => x.activo);
  const ordenados = activos.sort((a, b) => a.nombre.localeCompare(b.nombre));
  this.items.set(ordenados);
});
// DESPUÉS: pipeline declarativo, el subscribe solo asigna
source$.pipe(
  map(data => data.filter(x => x.activo)),
  map(activos => [...activos].sort((a, b) => a.nombre.localeCompare(b.nombre)))
).subscribe(items => this.items.set(items));

BehaviorSubject como store cuando computed alcanza

Un BehaviorSubject para modelar estado derivado es arrastrar RxJS a un problema de signals. Si el valor es “el estado actual, derivado de otro estado”, es un computed:

// ANTES (antipatrón): estado con BehaviorSubject y derivación manual
private items$ = new BehaviorSubject<Item[]>([]);
private filtro$ = new BehaviorSubject<string>('');
readonly visibles$ = combineLatest([this.items$, this.filtro$]).pipe(
  map(([items, f]) => items.filter(i => i.nombre.includes(f)))
);
// DESPUÉS (mejor): signals + computed, síncrono y sin suscripciones
readonly items = signal<Item[]>([]);
readonly filtro = signal('');
readonly visibles = computed(() =>
  this.items().filter(i => i.nombre.includes(this.filtro()))
);

El computed es pull-based y memoizado: solo recalcula cuando lo leen y cambió una dependencia; el combineLatest recalcula en cada emisión aunque nadie escuche. Reservá BehaviorSubject para cuando de verdad necesitás el eje temporal (multicast de eventos, historial de emisiones).

Otros a vigilar en revisión

Checklist

Ya sabés cuándo la reactividad debe ser un stream y cuándo un valor. El próximo paso es entender cómo Angular decide re-renderizar ese valor: cómo funciona la detección de cambios, por qué OnPush y zoneless cambian las reglas, y cómo medir el costo real de render. Seguimos en Change Detection y rendimiento de runtime →.

← Volver al listado de tecnologías ← Índice de Angular: Buenas Prácticas