RxJs: Operadores de combinación de observables

Cuando una aplicación reactiva necesita coordinar eventos de teclado, clics, respuestas HTTP y temporizadores, el reto no está solo en crear observables, sino en decidir cómo sincronizar sus emisiones. Combinar observables RxJS correctamente evita acoplamiento innecesario, carreras difíciles de depurar y estados intermedios inconsistentes en la interfaz.

En este artículo analizaremos operadores y funciones de combinación de RxJS: startWith, endWith, concat, merge, combineLatest y forkJoin. Cada uno resuelve un problema distinto: inicializar un stream, añadir una emisión final, concatenar secuencias, mezclar eventos concurrentes, sincronizar últimos valores o esperar a que varias fuentes finalicen.

startWith: inicializar un observable con una primera emisión

El operador startWith permite definir una o varias emisiones iniciales antes de que el observable fuente empiece a entregar sus propios valores. Es útil cuando una suscripción necesita un estado inicial inmediato, por ejemplo, para pintar una vista antes de que ocurra el primer evento real.

En el diagrama, el valor a se emite antes que cualquier valor proveniente del observable original. La suscripción no necesita esperar a que la fuente produzca el primer dato.

Ejemplo de uso:

import { of } from 'rxjs';
import { startWith } from 'rxjs/operators';

const numeros$ = of(1, 2, 3).pipe(
  startWith(3)
);

numeros$.subscribe(console.log);

Este flujo primero emite 3 por efecto de startWith(3) y después continúa con las emisiones de of(1, 2, 3). En escenarios de UI, este patrón suele aparecer cuando necesitas mostrar un valor por defecto antes de recibir eventos del usuario o datos remotos.

endWith: añadir emisiones al completar el stream

El operador endWith trabaja en el extremo opuesto de startWith. En lugar de emitir valores al inicio, agrega emisiones al final, justo cuando el observable fuente completa.

Esto implica una condición importante: la fuente debe completar. Si el observable nunca envía la señal de complete, endWith no tendrá oportunidad de emitir sus valores finales.

El modo de uso puede combinar ambos operadores:

import { of } from 'rxjs';
import { startWith, endWith } from 'rxjs/operators';

const numeros$ = of(1, 2, 3).pipe(
  startWith('a', 'b', 'c'),
  endWith('x', 'y', 'z'),
);

numeros$.subscribe(console.log);

El resultado incluye las emisiones iniciales, los valores originales y las emisiones finales:

endWith es apropiado para flujos finitos donde quieres señalar un cierre lógico, adjuntar un marcador final o emitir un estado de terminación después de procesar una secuencia.

concat: combinar observables RxJS de forma secuencial

La función concat concatena observables uno detrás de otro. Se suscribe al primer observable, espera a que complete y solo entonces avanza al siguiente. El resultado es un observable nuevo que conserva el orden secuencial de las fuentes.

El diagrama muestra que las emisiones del segundo stream no empiezan hasta que el primero completa. Esta propiedad es crítica cuando el orden tiene semántica de negocio, por ejemplo, inicializar una configuración antes de consumir un flujo dependiente.

Su modo de uso es:

import { concat, interval } from 'rxjs';
import { take } from 'rxjs/operators';

const intervalo$ = interval(1000);

concat(
  intervalo$.pipe(take(2)),
  intervalo$.pipe(take(3)),
  [1, 2, 3, 5] // un iterable u objeto compatible con ObservableInput
).subscribe(console.log);

El primer intervalo$.pipe(take(2)) emite dos valores y completa. Después se ejecuta el segundo intervalo$.pipe(take(3)). Finalmente, concat consume el arreglo [1, 2, 3, 5] como una fuente compatible con ObservableInput.

El resultado respeta ese orden:

La restricción operativa más importante es que cada fuente debe completar para que concat avance. Si el primer observable es infinito, el segundo nunca se suscribirá.

merge: combinar emisiones concurrentes en un solo observable

La función merge combina las emisiones de dos o más observables en un único stream de salida. A diferencia de concat, no espera a que una fuente complete para suscribirse a la siguiente; todas las fuentes activas pueden emitir de forma concurrente.

merge completa únicamente cuando todos los observables de entrada han completado. En streams de eventos del DOM, como click o keyup, esto normalmente significa que el flujo permanece activo durante toda la vida de la página, salvo que se aplique una estrategia de finalización explícita.

Modo de uso:

import { fromEvent, merge } from 'rxjs';
import { pluck } from 'rxjs/operators';

const keyup$ = fromEvent(document, 'keyup');
const click$ = fromEvent(document, 'click');

merge(
  keyup$.pipe(pluck('type')),
  click$.pipe(pluck('type'))
).subscribe(console.log);

En este caso, ambos observables se transforman para emitir únicamente el tipo de evento. El stream resultante recibe tanto keyup como click, sin imponer un orden artificial entre ellos.

El resultado en consola evidencia que ambos eventos se combinan en una sola suscripción:

combineLatest: sincronizar los últimos valores emitidos

La función combineLatest combina múltiples observables y emite un arreglo con el último valor disponible de cada fuente. Su comportamiento tiene una regla clave: no emite hasta que todos los observables participantes hayan emitido al menos una vez.

Después de esa primera sincronización, cada nueva emisión de cualquier fuente produce una nueva salida con el valor actualizado de esa fuente y los últimos valores conocidos de las demás. Esto es especialmente útil para formularios, filtros, coordenadas de UI y estados derivados.

El código de uso puede ser parecido a esto:

import { combineLatest, fromEvent } from 'rxjs';
import { pluck } from 'rxjs/operators';

const input1 = document.createElement('input');
const input2 = document.createElement('input');

input1.placeholder = 'email@example.com';

input2.placeholder = '*******';
input2.type = 'password';

document.querySelector('body')?.append(input1, input2);

// helper
const getInputString = (elem: HTMLElement) => {
  return fromEvent<KeyboardEvent>(elem, 'keyup')
    .pipe(
      pluck<KeyboardEvent>('target', 'value')
    );
};

combineLatest(
  getInputString(input1),
  getInputString(input2)
).subscribe(console.log);

El ejemplo conserva la firma posicional usada durante mucho tiempo en RxJS. En código moderno, especialmente con RxJS 7+, suele preferirse la firma con arreglo porque escala mejor y es más explícita:

combineLatest([
  getInputString(input1),
  getInputString(input2),
]).subscribe(console.log);

También conviene considerar que pluck pertenece a un estilo más antiguo de transformación. Para proyectos nuevos, map con acceso explícito suele ser más legible y compatible con la evolución del ecosistema.

forkJoin: esperar la última emisión de observables finitos

La función forkJoin emite una sola vez con la última emisión de cada observable de entrada, pero únicamente cuando todos completan. Este comportamiento la convierte en una herramienta precisa para coordinar operaciones finitas, como peticiones HTTP simultáneas.

El diagrama muestra que la salida aparece al final, cuando todas las fuentes han completado y existe un último valor para cada una.

Modo de uso:

import { of, interval, forkJoin } from 'rxjs';
import { take, delay } from 'rxjs/operators';

const numeros$ = of(1, 2, 3, 4);
const intervalo$ = interval(1000).pipe(take(3));
const letras$ = of('a', 'b', 'c').pipe(delay(3500));

// forkJoin(
//   numeros$,
//   intervalo$,
//   letras$
// ).subscribe(console.log);

// forkJoin(
//   numeros$,
//   intervalo$,
//   letras$
// ).subscribe(resp => {
//   console.log('numeros', resp[0]);
//   console.log('intervalo', resp[1]);
//   console.log('letras', resp[2]);
// });

// forkJoin({
//   numeros$,
//   intervalo$,
//   letras$
// }).subscribe(resp => {
//   console.log(resp);
// });

forkJoin({
  num: numeros$,
  int: intervalo$,
  let: letras$
}).subscribe(resp => {
  console.log(resp);
});

La forma con objeto es especialmente práctica porque conserva nombres semánticos en la respuesta. En lugar de depender de índices como resp[0], resp[1] o resp[2], el consumidor trabaja con propiedades explícitas: num, int y let.

Cómo elegir entre concat, merge, combineLatest y forkJoin

Estos operadores pueden parecer similares porque todos combinan fuentes, pero su semántica de tiempo es distinta:

• Usa concat cuando el orden secuencial sea obligatorio y cada fuente deba completar antes de iniciar la siguiente.
• Usa merge cuando las fuentes sean independientes y quieras reaccionar a cualquiera de ellas tan pronto como emita.
• Usa combineLatest cuando necesites calcular estado derivado a partir del último valor conocido de varias fuentes activas.
• Usa forkJoin cuando esperes observables finitos y solo necesites el resultado final de cada uno.
• Usa startWith y endWith para insertar emisiones de inicio o cierre dentro de un pipeline.

Esta elección tiene impacto directo en latencia, orden de ejecución y condiciones de completado. En sistemas reactivos, estos detalles definen si el flujo se mantiene responsivo o si queda bloqueado esperando una emisión que nunca llega.

Conclusión

Combinar observables RxJS requiere distinguir entre concurrencia, secuencialidad y sincronización de valores. concat serializa fuentes, merge las mezcla, combineLatest deriva estado desde los últimos valores y forkJoin espera resultados finales de observables finitos.

Dominar estas diferencias permite diseñar pipelines más predecibles y expresivos. En el siguiente artículo aplicaremos estos conceptos al trabajo con peticiones AJAX, donde forkJoin, mergeMap y otros operadores empiezan a interactuar con latencia real de red y manejo de errores.