RxJs: Operadores no tan comunes

En flujos reactivos de interfaz, temporizadores o eventos del DOM, no siempre necesitas transformar cada emisión. Con frecuencia el problema operativo consiste en limitar una suscripción, cancelar un Observable ante una señal externa o eliminar emisiones repetidas. Estos operadores RxJS resuelven ese tipo de control sin introducir estado mutable adicional ni lógica imperativa alrededor de la suscripción.

La referencia formal de estos operadores está en la documentación oficial de RxJS y en el modelo conceptual de ReactiveX. En este artículo nos concentraremos en su comportamiento práctico, especialmente en cuándo completan el flujo y cómo comparan valores.

take: limitar emisiones de un Observable

take permite definir cuántas emisiones del Observable fuente queremos recibir. Una vez alcanza ese límite, el operador completa la cadena y cancela la suscripción aguas arriba.

Este patrón es útil cuando el consumidor solo necesita una muestra finita de un flujo potencialmente largo, como los primeros eventos, los primeros valores de un intervalo o una respuesta inicial.

import { of } from "rxjs";
import { take, tap } from "rxjs/operators";

const numeros$ = of(1, 2, 3, 4, 5);

numeros$
  .pipe(
    tap((valor) => console.log("tap", valor)),
    take(3),
  )
  .subscribe({
    next: (valor) => console.log("next", valor),
    complete: () => console.log("complete"),
  });

El tap se ejecuta solo para los valores que alcanzan la suscripción antes de que take(3) complete el flujo.

first: tomar la primera emisión o la primera que cumpla una condición

first obtiene la primera emisión del Observable y completa la suscripción. Su caso básico se parece a take(1), pero también acepta un predicado para esperar la primera emisión que cumpla una condición.

import { fromEvent } from "rxjs";
import { first } from "rxjs/operators";

const click$ = fromEvent<PointerEvent>(document, "click");

click$.pipe(first()).subscribe({
  next: (event) => console.log("next", event),
  complete: () => console.log("complete"),
});

Después del primer clic, el flujo completa y deja de escuchar nuevos eventos.

first con predicado

Cuando first recibe un predicado, ignora las emisiones previas hasta encontrar la primera que devuelva true.

import { fromEvent } from "rxjs";
import { first, tap } from "rxjs/operators";

const click$ = fromEvent<PointerEvent>(document, "click");

click$
  .pipe(
    tap(() => console.log("tap")),
    first((event) => event.clientY >= 150),
  )
  .subscribe({
    next: (event) => console.log("next", event),
    complete: () => console.log("complete"),
  });

El tap confirma que el flujo sigue recibiendo clics, pero first solo emite cuando la coordenada clientY cumple la condición.

takeWhile: emitir mientras se mantenga una condición

takeWhile mantiene la suscripción activa mientras el predicado devuelva true. En cuanto una emisión no cumple la condición, el operador completa el flujo.

En este ejemplo tomamos clics mientras la coordenada y sea menor o igual a 150.

import { fromEvent } from "rxjs";
import { map, takeWhile } from "rxjs/operators";

const click$ = fromEvent<PointerEvent>(document, "click");

click$
  .pipe(
    map(({ x, y }) => ({ x, y })),
    takeWhile(({ y }) => y <= 150, true),
  )
  .subscribe({
    next: (coords) => console.log("next", coords),
    complete: () => console.log("complete"),
  });

El segundo argumento, conocido como bandera inclusive, indica si la emisión que rompe la condición también debe enviarse al suscriptor. Por defecto es false; con true, el valor que provoca el cierre se conserva.

takeUntil: cancelar un flujo con otro Observable

takeUntil recibe un segundo Observable, llamado normalmente notifier. El flujo principal emite hasta que ese notifier produce su primera emisión; en ese momento la cadena completa.

Este patrón es común para cancelar intervalos, peticiones, listeners o cualquier proceso reactivo asociado al ciclo de vida de un componente.

import { fromEvent, interval } from "rxjs";
import { takeUntil } from "rxjs/operators";

const boton = document.createElement("button");
boton.innerHTML = "Detener timing";

document.querySelector("body")?.append(boton);

const counter$ = interval(1000);
const click$ = fromEvent(boton, "click");

counter$.pipe(takeUntil(click$)).subscribe({
  next: (valor) => console.log("next", valor),
  complete: () => console.log("complete"),
});

El contador sigue emitiendo cada segundo hasta que el botón produce un evento click.

skip: omitir emisiones iniciales

skip ignora las primeras n emisiones del Observable fuente y deja pasar las posteriores. Es útil cuando las primeras señales representan inicialización, ruido operativo o eventos que no deben activar la lógica principal.

Aplicado al ejemplo anterior, podemos exigir varios clics antes de cancelar el contador. Con skip(2), el tercer clic es el primero que llega a takeUntil.

import { fromEvent, interval } from "rxjs";
import { skip, takeUntil, tap } from "rxjs/operators";

const boton = document.createElement("button");
boton.innerHTML = "Detener timing";

document.querySelector("body")?.append(boton);

const counter$ = interval(1000);
const click$ = fromEvent(boton, "click").pipe(
  tap(() => console.log("tap-antes")),
  skip(2),
  tap(() => console.log("tap-despues")),
);

counter$.pipe(takeUntil(click$)).subscribe({
  next: (valor) => console.log("next", valor),
  complete: () => console.log("complete"),
});

La salida muestra que los primeros clics existen, pero no cruzan la frontera marcada por skip.

distinct: eliminar valores repetidos globalmente

distinct evita que el flujo emita valores ya observados previamente. La comparación base usa igualdad estricta (===), por lo que tipo y valor deben coincidir.

import { of } from "rxjs";
import { distinct } from "rxjs/operators";

const numeros$ = of(1, 1, 1, 3, 3, 2, 2, 4, 5, 3, 1);

numeros$.pipe(distinct()).subscribe(console.log);

El operador conserva una memoria interna de valores ya emitidos y descarta repeticiones posteriores.

distinct con selector para objetos

Con objetos, dos instancias con la misma estructura no son iguales por referencia. Para resolverlo, distinct acepta un selector que define la clave de comparación.

import { from } from "rxjs";
import { distinct } from "rxjs/operators";

interface Personaje {
  nombre: string;
}

const personajes: Personaje[] = [
  { nombre: "Megaman" },
  { nombre: "Zero" },
  { nombre: "X" },
  { nombre: "Zero" },
  { nombre: "Zero" },
  { nombre: "Zero" },
  { nombre: "X" },
  { nombre: "Megaman" },
  { nombre: "Megaman" },
];

from(personajes)
  .pipe(distinct((personaje) => personaje.nombre))
  .subscribe(console.log);

En este caso la unicidad se calcula por nombre, no por referencia del objeto.

distinctUntilChanged: evitar repeticiones consecutivas

distinctUntilChanged compara cada emisión con la emisión inmediatamente anterior. Solo descarta duplicados consecutivos; si un valor reaparece más adelante después de otro valor distinto, puede emitirse de nuevo.

import { of } from "rxjs";
import { distinctUntilChanged } from "rxjs/operators";

const numeros$ = of(1, "1", 1, 1, 3, 3, 2, 2, 2, 4, 5, 3, 1);

numeros$.pipe(distinctUntilChanged()).subscribe(console.log);

La igualdad por defecto también usa ===; por eso 1 y "1" no se consideran duplicados.

distinctUntilChanged con comparador

Para objetos, el comparador recibe el valor anterior y el actual. Debe retornar true cuando ambos se consideran equivalentes y, por tanto, el valor actual debe omitirse.

import { from } from "rxjs";
import { distinctUntilChanged } from "rxjs/operators";

interface Personaje {
  nombre: string;
}

const personajes: Personaje[] = [
  { nombre: "Megaman" },
  { nombre: "Megaman" },
  { nombre: "Zero" },
  { nombre: "X" },
  { nombre: "Zero" },
  { nombre: "Zero" },
  { nombre: "Zero" },
  { nombre: "X" },
  { nombre: "Megaman" },
  { nombre: "Megaman" },
];

from(personajes)
  .pipe(
    distinctUntilChanged(
      (anterior, actual) => anterior.nombre === actual.nombre,
    ),
  )
  .subscribe(console.log);

El flujo descarta únicamente los personajes consecutivos con el mismo nombre.

distinctUntilKeyChanged: comparación consecutiva por propiedad

distinctUntilKeyChanged aplica la misma idea de distinctUntilChanged, pero optimizada para objetos cuando la comparación depende de una propiedad concreta.

import { from } from "rxjs";
import { distinctUntilKeyChanged } from "rxjs/operators";

interface Personaje {
  nombre: string;
}

const personajes: Personaje[] = [
  { nombre: "Megaman" },
  { nombre: "Megaman" },
  { nombre: "Zero" },
  { nombre: "X" },
  { nombre: "Zero" },
  { nombre: "Zero" },
  { nombre: "Zero" },
  { nombre: "X" },
  { nombre: "Megaman" },
  { nombre: "Megaman" },
];

from(personajes)
  .pipe(distinctUntilKeyChanged("nombre"))
  .subscribe(console.log);

La intención queda más explícita que con un comparador manual cuando solo necesitas evaluar una clave.

Conclusión

Estos operadores no son secundarios; resuelven decisiones críticas de ciclo de vida, cancelación y deduplicación en flujos reactivos. take, first, takeWhile, takeUntil y skip controlan cuántas emisiones llegan al consumidor y cuándo termina la suscripción. distinct, distinctUntilChanged y distinctUntilKeyChanged controlan repetición y semántica de igualdad.

La regla práctica es seleccionar el operador según el límite que necesitas expresar: cantidad, condición, señal externa, emisiones iniciales o duplicados. Esa precisión mantiene tus pipelines de RxJS más declarativos, auditables y fáciles de mantener.