RxJs: Operadores de transformación

Cuando una aplicación combina eventos de UI, peticiones HTTP y procesos asíncronos, el flujo deja de ser una secuencia plana de valores y empieza a producir observables internos. Los operadores de transformación RxJS resuelven ese punto crítico: convierten un observable de observables en un flujo consumible, controlando concurrencia, cancelación, orden y exclusión de trabajo duplicado.

La documentación oficial de RxJS describe estos operadores como herramientas para trabajar con higher-order observables: observables cuyas emisiones son, a su vez, otros observables. En términos prácticos, aparecen cada vez que un evento dispara trabajo asíncrono: una búsqueda, un guardado, una carga incremental o una operación remota.

La diferencia entre estos operadores no es cosmética. Elegir uno u otro define cómo se comporta el sistema ante emisiones rápidas: si ejecuta todo en paralelo, cancela lo anterior, serializa las tareas o ignora nuevos eventos mientras una operación sigue activa.

mergeAll: aplanar observables internos con concurrencia

mergeAll trabaja sobre un observable que emite observables. Se suscribe a cada observable interno y reemite sus valores en el flujo resultante. El resultado se completa únicamente cuando el observable exterior se completa y todos los observables internos activos también han completado.

Este operador es útil cuando ya tienes construidos los observables internos y solo necesitas aplanarlos. Por ejemplo, una capa previa puede generar varias operaciones independientes: guardar un borrador, registrar auditoría y sincronizar métricas. Si ninguna depende del resultado de la otra, mergeAll permite ejecutarlas con concurrencia.

En el diagrama, cada emisión del flujo superior representa un observable interno. mergeAll mezcla sus emisiones conforme llegan, sin preservar necesariamente el orden original entre observables.

import { from, mapTo, mergeAll, timer } from 'rxjs';

const saveDraft$ = timer(300).pipe(mapTo('draft saved'));
const sendAudit$ = timer(100).pipe(mapTo('audit sent'));
const syncMetrics$ = timer(200).pipe(mapTo('metrics synced'));

from([saveDraft$, sendAudit$, syncMetrics$])
  .pipe(mergeAll())
  .subscribe(console.log);

// audit sent
// metrics synced
// draft saved

El ejemplo muestra un comportamiento clave: el orden de salida depende del tiempo de finalización de cada observable interno. Si necesitas preservar orden estricto, mergeAll no es la herramienta adecuada; concatMap encaja mejor en ese caso.

mergeMap: transformar y mezclar con concurrencia completa

mergeMap combina dos pasos: transforma cada emisión del observable fuente en un observable interno y luego mezcla las emisiones de todos esos observables. Conceptualmente equivale a aplicar map para crear observables y después mergeAll para aplanarlos.

Su semántica favorece la concurrencia. Cada emisión entrante dispara una nueva suscripción interna, y las suscripciones anteriores permanecen activas hasta completar. Esto lo hace adecuado para operaciones independientes donde cancelar trabajo previo sería incorrecto, como escrituras, telemetría, cargas paralelas o procesamiento de elementos sin dependencia de orden.

import { from, mergeMap } from 'rxjs';
import { ajax } from 'rxjs/ajax';

interface User {
  id: number;
  name: string;
}

const userIds = [101, 102, 103, 104];

from(userIds)
  .pipe(
    mergeMap(
      (id) => ajax.getJSON<User>(`/api/users/${id}`),
      4 // límite de concurrencia
    )
  )
  .subscribe((user) => {
    console.log('Usuario cargado:', user.name);
  });

Aunque mergeMap suele asociarse con “concurrencia completa”, RxJS permite limitar esa concurrencia con el segundo argumento. Este detalle es importante cuando el flujo fuente puede emitir muchos valores y cada valor dispara trabajo costoso, como peticiones HTTP o acceso a recursos externos.

switchMap: cancelar lo anterior y conservar lo más reciente

switchMap también transforma cada emisión en un observable interno, pero mantiene activa solo la suscripción más reciente. Cuando llega una nueva emisión desde el observable fuente, cancela la suscripción interna anterior y se cambia al nuevo observable.

Esta semántica es ideal para escenarios donde el resultado anterior pierde relevancia al aparecer una entrada más nueva. El caso clásico es un typeahead o autocomplete: si el usuario escribe rea y luego react, la respuesta de rea ya no debe actualizar la UI aunque llegue después.

import {
  debounceTime,
  distinctUntilChanged,
  fromEvent,
  map,
  of,
  switchMap,
} from 'rxjs';
import { ajax } from 'rxjs/ajax';

interface SearchResult {
  id: string;
  title: string;
}

const input = document.querySelector<HTMLInputElement>('#search')!;

fromEvent<InputEvent>(input, 'input')
  .pipe(
    map((event) => (event.target as HTMLInputElement).value.trim()),
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((term) =>
      term.length === 0
        ? of([])
        : ajax.getJSON<SearchResult[]>(`/api/search?q=${encodeURIComponent(term)}`)
    )
  )
  .subscribe((results) => {
    console.log('Resultados vigentes:', results);
  });

El valor operativo de switchMap está en la cancelación. En Angular, por ejemplo, cancelar la suscripción a una petición HTTP puede abortar la solicitud subyacente; en otros contextos, al menos evita que respuestas obsoletas lleguen al consumidor final.

concatMap: preservar orden y ejecutar de forma secuencial

concatMap transforma cada emisión en un observable interno, pero no se suscribe al siguiente hasta que el observable interno actual completa. En lugar de mezclar resultados, crea una cola: procesa una tarea, espera su finalización y continúa con la siguiente.

Este comportamiento es crucial cuando el orden importa o cuando el backend espera operaciones serializadas. Casos típicos incluyen guardar cambios en secuencia, ejecutar comandos de dominio, procesar pasos de onboarding o enviar eventos que deben respetar causalidad.

import { concatMap, from } from 'rxjs';
import { ajax } from 'rxjs/ajax';

interface OrderCommand {
  orderId: string;
  status: 'validated' | 'charged' | 'confirmed';
}

const commands: OrderCommand[] = [
  { orderId: 'A-100', status: 'validated' },
  { orderId: 'A-100', status: 'charged' },
  { orderId: 'A-100', status: 'confirmed' },
];

from(commands)
  .pipe(
    concatMap((command) =>
      ajax.post('/api/orders/commands', command, {
        'Content-Type': 'application/json',
      })
    )
  )
  .subscribe(() => {
    console.log('Comando procesado en orden');
  });

La principal ventaja de concatMap es la predictibilidad. La desventaja es la latencia acumulada: si cada operación tarda 500 ms y tienes diez elementos, el flujo tardará al menos cinco segundos en procesar la cola completa.

exhaustMap: ignorar nuevas emisiones mientras hay trabajo activo

exhaustMap transforma una emisión en un observable interno y, mientras ese observable está activo, ignora todas las emisiones posteriores del observable fuente. Cuando el observable interno completa, el operador queda listo para aceptar una nueva emisión.

Este patrón es útil cuando una operación no debe ejecutarse dos veces en paralelo. Formularios de login, botones de pago, acciones de envío y procesos idempotentes costosos suelen beneficiarse de esta semántica: el primer evento inicia el trabajo y los siguientes clics se descartan hasta terminar.

import { exhaustMap, fromEvent } from 'rxjs';
import { ajax } from 'rxjs/ajax';

interface PaymentResult {
  transactionId: string;
}

const payButton = document.querySelector<HTMLButtonElement>('#pay')!;
const payload = { orderId: 'A-100' };

fromEvent(payButton, 'click')
  .pipe(
    exhaustMap(() =>
      ajax.post<PaymentResult>('/api/payments', payload, {
        'Content-Type': 'application/json',
      })
    )
  )
  .subscribe((response) => {
    console.log('Pago procesado:', response.response.transactionId);
  });

A diferencia de switchMap, exhaustMap no cancela el trabajo activo. A diferencia de mergeMap, tampoco crea trabajo adicional. Su contrato es deliberadamente estricto: “ignorar mientras esté ocupado”.

Cuándo elegir cada operador de transformación RxJS

La elección correcta depende de la política de concurrencia que necesita el sistema, no solo de la forma del código.

• Usa mergeAll cuando recibes observables internos ya creados y quieres mezclarlos.
• Usa mergeMap cuando cada emisión dispara trabajo independiente y aceptas concurrencia.
• Usa switchMap cuando solo importa el resultado más reciente y conviene cancelar lo anterior.
• Usa concatMap cuando necesitas orden, serialización o consistencia secuencial.
• Usa exhaustMap cuando una operación activa debe bloquear nuevas emisiones hasta completar.

Una regla práctica es observar qué debería ocurrir si llegan tres emisiones rápidas. Si las tres deben ejecutarse, mergeMap. Si solo la última importa, switchMap. Si las tres deben ejecutarse en orden, concatMap. Si solo la primera debe ejecutarse y las demás deben ignorarse mientras está activa, exhaustMap.

Conclusión

Los operadores de transformación RxJS definen cómo se conectan eventos de alto nivel con trabajo asíncrono real. mergeAll, mergeMap, switchMap, concatMap y exhaustMap no son variantes intercambiables; cada uno codifica una política distinta de concurrencia, cancelación y orden.

Al diseñar flujos reactivos en TypeScript, conviene elegir el operador a partir del contrato del dominio. Esa decisión reduce condiciones de carrera, evita respuestas obsoletas, controla presión sobre recursos externos y hace que la intención del flujo sea explícita para el equipo.