RxJs: Observables, Subscriptions y Subjects

Cuando un flujo asíncrono combina eventos de usuario, timers, llamadas HTTP y actualizaciones de estado, el problema no es solo recibir datos: es modelar correctamente la vida útil de cada stream. Observables RxJS permite representar esos flujos como secuencias cancelables, componibles y explícitas, con una semántica clara para emisión, error, finalización y limpieza de recursos.

Crear observables rxjs con new Observable

En RxJS, un Observable representa una fuente de valores que puede emitir cero, uno o muchos elementos a lo largo del tiempo. La forma más directa de construir uno es instanciar Observable y entregar una función de suscripción. Esa función recibe un subscriber, que actúa como canal para emitir valores con next, reportar errores con error o finalizar con complete.

import { Observable } from 'rxjs';

const obs$ = new Observable<string>((subscriber) => {
  subscriber.next('Hola');
  subscriber.next('mundo');

  subscriber.next('Hola');
  subscriber.next('mundo');

  subscriber.complete();

  // Después de complete(), estas emisiones ya no llegan al consumidor.
  subscriber.next('Hola');
  subscriber.next('mundo');
});

La convención del sufijo $ en obs$ no es obligatoria, pero ayuda a comunicar que una variable representa un stream. También es importante declarar el tipo del observable, por ejemplo Observable<string>, porque TypeScript podrá validar el contrato entre la fuente y sus consumidores.

complete() no destruye el objeto observable; cierra la ejecución de esa suscripción específica. A partir de ese punto, el consumidor ya no recibe nuevas emisiones de esa instancia de ejecución.

La interfaz Observer: next, error y complete

Para consumir un observable usamos subscribe. En versiones modernas de RxJS es preferible pasar un objeto compatible con la interfaz Observer, en lugar de usar la sobrecarga antigua de tres callbacks separados.

import { Observable, Observer } from 'rxjs';

const observer: Observer<string> = {
  next: (valor) => {
    console.log('siguiente', valor);
  },
  error: (err) => {
    console.error('error', err);
  },
  complete: () => {
    console.info('completado');
  },
};

const obs$ = new Observable<string>((subscriber) => {
  subscriber.next('Hola');
  subscriber.next('mundo');
  subscriber.complete();
});

obs$.subscribe(observer);

La interfaz divide explícitamente los tres canales del stream:

• next: se ejecuta cada vez que el observable emite un valor.
• error: se ejecuta cuando la fuente reporta un fallo terminal.
• complete: se ejecuta cuando la fuente finaliza correctamente.

error y complete son señales terminales. Después de cualquiera de ellas, la suscripción queda cerrada y no debe esperarse más trabajo desde esa ejecución.

// Forma antigua: evita esta sobrecarga en código nuevo.
obs$.subscribe(
  (valor) => console.log('Next', valor),
  (err) => console.error('Error', err),
  () => console.info('complete'),
);

// Forma recomendada.
obs$.subscribe(observer);

La documentación oficial de ReactiveX describe este contrato como una secuencia de notificaciones donde el observable invoca al observer. RxJS conserva esa idea y la adapta al ecosistema JavaScript con cancelación explícita mediante suscripciones.

Subscription y unsubscribe: controlar la vida útil del stream

Cada llamada a subscribe crea una Subscription. Esa suscripción representa la ejecución activa entre un observable y un consumidor. Si el observable administra recursos externos, como un setInterval, un WebSocket o un listener del DOM, la suscripción debe liberar esos recursos cuando el consumidor ya no los necesita.

import { Observable } from 'rxjs';

const intervalos$ = new Observable<number>((subscriber) => {
  let contador = 0;

  const intervalId = setInterval(() => {
    contador++;
    subscriber.next(contador);
  }, 1000);

  return () => {
    clearInterval(intervalId);
    console.log('Intervalo destruido');
  };
});

const subs1 = intervalos$.subscribe((valor) => console.log('subs1:', valor));
const subs2 = intervalos$.subscribe((valor) => console.log('subs2:', valor));
const subs3 = intervalos$.subscribe((valor) => console.log('subs3:', valor));

setTimeout(() => {
  subs1.unsubscribe();
  subs2.unsubscribe();
  subs3.unsubscribe();

  console.log('Completado Timeout');
}, 3000);

En este ejemplo, cada suscripción recibe su propio contador porque el observable es cold: la producción de datos ocurre dentro de la función de suscripción. Tres suscripciones implican tres ejecuciones independientes del setInterval.

La función retornada desde el constructor del observable es el teardown callback. RxJS la ejecuta cuando llamamos unsubscribe() o cuando la suscripción se cierra por una señal terminal. Es el lugar correcto para liberar timers, conexiones, listeners o cualquier recurso asociado a esa ejecución.

Finalizar desde el observable con complete()

También es posible cerrar la suscripción desde la propia fuente. Para ello, el observable invoca subscriber.complete(). Cuando esto ocurre, RxJS ejecuta el callback complete del observer y posteriormente ejecuta el teardown registrado.

import { Observable, Observer } from 'rxjs';

const observer: Observer<number> = {
  next: (valor) => {
    console.log('siguiente', valor);
  },
  error: (err) => {
    console.error('error', err);
  },
  complete: () => {
    console.info('completado');
  },
};

const intervalos$ = new Observable<number>((subscriber) => {
  let contador = 0;

  const intervalId = setInterval(() => {
    contador++;
    subscriber.next(contador);
    console.log('obs:', contador);
  }, 1000);

  setTimeout(() => {
    subscriber.complete();
  }, 2500);

  return () => {
    clearInterval(intervalId);
    console.log('Intervalo destruido');
  };
});

const subs1 = intervalos$.subscribe(observer);
const subs2 = intervalos$.subscribe(observer);
const subs3 = intervalos$.subscribe(observer);

setTimeout(() => {
  subs1.unsubscribe();
  subs2.unsubscribe();
  subs3.unsubscribe();

  console.log('Completado Timeout');
}, 6000);

Aunque el setTimeout final intenta cancelar las suscripciones a los seis segundos, cada una ya fue completada por el observable a los 2.5 segundos. La llamada posterior a unsubscribe() no vuelve a emitir complete; simplemente opera sobre una suscripción cerrada.

Esta diferencia es relevante en diseño de APIs. complete() comunica que la fuente terminó correctamente; unsubscribe() comunica que el consumidor ya no quiere recibir valores. Ambas rutas cierran la suscripción, pero expresan intenciones distintas.

Encadenar suscripciones con .add()

Cuando varias suscripciones comparten el mismo ciclo de vida, RxJS permite agruparlas con .add(). Esto evita repetir llamadas a unsubscribe() y centraliza la limpieza en una suscripción raíz.

const subs1 = intervalos$.subscribe(observer);
const subs2 = intervalos$.subscribe(observer);
const subs3 = intervalos$.subscribe(observer);

subs1.add(subs2);
subs1.add(subs3);

setTimeout(() => {
  subs1.unsubscribe();

  console.log('Completado Timeout');
}, 6000);

Al cancelar subs1, RxJS cancela también las suscripciones agregadas. Este patrón resulta útil cuando un componente, servicio o caso de uso crea múltiples streams relacionados y necesita destruirlos como una unidad.

Subject RxJS: multicast y observable caliente

Un Subject combina dos roles: es un observable y también es un observer. Puede recibir valores mediante next, error o complete, y al mismo tiempo permite que múltiples consumidores se suscriban a él.

import { Observable, Subject } from 'rxjs';

const intervalo$ = new Observable<number>((subscriber) => {
  const intervalId = setInterval(() => {
    subscriber.next(Math.random());
  }, 3000);

  return () => clearInterval(intervalId);
});

const subject$ = new Subject<number>();

intervalo$.subscribe(subject$);

const subs1 = subject$.subscribe((num) => console.log('subs1:', num));
const subs2 = subject$.subscribe((num) => console.log('subs2:', num));

La diferencia clave es que Subject habilita multicast: una sola ejecución de la fuente puede distribuir la misma emisión a varios observers. En lugar de crear un setInterval por cada consumidor, el Subject actúa como punto de difusión.

flowchart LR
    A[Observable fuente] --> B[Subject]
    B --> C[Observer 1]
    B --> D[Observer 2]
    B --> E[Observer 3]

En términos de hot y cold observables, un observable es cold cuando la producción de datos ocurre dentro de la suscripción y cada consumidor obtiene su propia ejecución. Un observable es hot cuando la producción de datos existe fuera de una suscripción individual y puede compartirse entre varios consumidores.

El Subject puede convertir un flujo cold en un flujo hot desde la perspectiva de los subscribers conectados al subject. La fuente se ejecuta una vez y el Subject distribuye los valores a múltiples listeners.

Emitir y completar desde un Subject

Como Subject también es un observer, puede emitir valores propios. Esto permite introducir valores manualmente en el stream, además de reenviar los valores provenientes de una fuente.

import { Observable, Observer, Subject } from 'rxjs';

const observer: Observer<number> = {
  next: (valor) => {
    console.log('siguiente', valor);
  },
  error: (err) => {
    console.error('error', err);
  },
  complete: () => {
    console.info('completado');
  },
};

const intervalo$ = new Observable<number>((subscriber) => {
  const intervalId = setInterval(() => {
    subscriber.next(Math.random());
  }, 1000);

  return () => {
    clearInterval(intervalId);
    console.log('SetInterval terminado');
  };
});

const subject$ = new Subject<number>();
const baseInterval = intervalo$.subscribe(subject$);

const subs1 = subject$.subscribe(observer);
const subs2 = subject$.subscribe(observer);

setTimeout(() => {
  subject$.next(10);
  subject$.complete();

  baseInterval.unsubscribe();
}, 3500);

El detalle operativo es importante: llamar subject$.complete() cierra el subject y notifica a sus suscriptores, pero no necesariamente cancela la suscripción original que alimentaba al subject. Por eso el ejemplo ejecuta baseInterval.unsubscribe() de forma explícita.

Conclusión

Los observables rxjs ofrecen un contrato robusto para representar streams asíncronos en TypeScript. Observer define cómo reaccionar ante next, error y complete; Subscription controla la vida útil de la ejecución; unsubscribe() dispara la limpieza mediante el teardown callback; y .add() permite componer ciclos de vida relacionados.

Subject amplía el modelo al habilitar multicast y convertir flujos cold en flujos hot bajo ciertas condiciones. Esta capacidad es útil, pero exige ownership explícito: completar el subject y cancelar la fuente no siempre son la misma operación.

En el siguiente artículo abordaremos las funciones para crear Observables, una forma más declarativa y menos manual de construir streams con RxJS.