RxJs: Funciones para crear Observables

Cuando modelamos interacciones de usuario, temporizadores, colecciones o tareas diferidas, la decisión inicial suele ser cómo crear observables RxJS sin introducir acoplamiento innecesario ni perder claridad sobre la sincronía del flujo. En RxJS, las funciones de creación encapsulan patrones comunes para construir Observables a partir de valores, iterables, eventos del DOM y relojes internos.

Crear observables RxJS con funciones de creación

Las funciones de creación producen un Observable sin escribir manualmente el constructor new Observable(...). Esto reduce código repetitivo y comunica la intención del stream: emitir valores estáticos, recorrer una colección, escuchar eventos o programar emisiones en el tiempo.

La diferencia más importante está en la naturaleza síncrona o asíncrona del observable resultante. of, from y range pueden emitir inmediatamente durante la suscripción. fromEvent, interval y timer, en cambio, dependen de eventos externos o del paso del tiempo.

of: emitir cada argumento como un valor independiente

of crea un observable síncrono a partir de los argumentos que recibe. Cada argumento se convierte en una emisión individual y, al terminar la lista, el stream envía complete.

Esta característica es útil para valores estáticos, pruebas unitarias, mocks de servicios o flujos donde necesitas transformar una secuencia pequeña sin depender de una fuente externa.

import { of } from 'rxjs';

const obs$ = of(1, 'string', true, () => {}, 1.14);

console.log('Inicio del obs$');
obs$.subscribe({
  next: value => console.log('next:', value),
  complete: () => console.log('Completado!'),
});
console.log('Fin del obs$');

Como of es síncrono, todas las emisiones ocurren antes de ejecutar console.log('Fin del obs$'). Esta propiedad es relevante cuando mezclas streams síncronos con lógica de UI o con efectos secundarios que dependen del orden de ejecución.

from: convertir iterables, arreglos y promesas en Observables

from crea un observable a partir de estructuras que ya representan una secuencia o un valor diferido: arreglos, iterables, strings, Promise, entre otros. Con un arreglo, emite cada elemento de manera independiente y luego completa.

import { from } from 'rxjs';

interface Personaje {
  tipo: 'heroe' | 'villano';
  nombre: string;
}

const personajes: Personaje[] = [
  { tipo: 'heroe', nombre: 'Batman' },
  { tipo: 'heroe', nombre: 'Robin' },
  { tipo: 'villano', nombre: 'Joker' },
];

from(personajes).subscribe(personaje => console.log('from', personaje));

La diferencia conceptual entre of(personajes) y from(personajes) es crítica. El primero emite un solo arreglo; el segundo emite tres objetos. Esa elección define el contrato del stream para los operadores posteriores.

fromEvent: crear Observables desde eventos del DOM

fromEvent convierte una fuente de eventos en un observable. Recibe el target que expone eventos —por ejemplo document, window o un elemento HTML— y el nombre del evento.

import { fromEvent } from 'rxjs';

const clicks$ = fromEvent<PointerEvent>(document, 'click');
const keyups$ = fromEvent<KeyboardEvent>(document, 'keyup');

clicks$.subscribe(({ x, y }) => console.log('click:', x, y));
keyups$.subscribe(event => console.log('key:', event.key));

Este observable es infinito por defecto. El navegador seguirá emitiendo eventos mientras exista la suscripción, por lo que en componentes con ciclo de vida —por ejemplo Angular, React o Web Components— debes liberar recursos cuando el componente se desmonta.

Identificar el tipo del evento sin usar any

En TypeScript conviene evitar fromEvent<any>, porque rompe el contrato de tipos justo en la frontera del sistema. Una forma práctica de identificar el tipo consiste en inspeccionar temporalmente el evento y reemplazar any por la interfaz específica.

import { fromEvent } from 'rxjs';

const source$ = fromEvent<any>(document, 'click');
source$.subscribe(console.log);

Al revisar el objeto emitido en consola o desde el editor, puedes identificar si corresponde a PointerEvent, KeyboardEvent, MouseEvent, InputEvent u otro tipo del DOM. Después reemplazas any por el tipo correcto.

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

source$.subscribe(({ x, y }) => {
  console.log({ x, y });
});

range: generar secuencias numéricas

range crea una secuencia numérica finita. Por defecto emite de forma síncrona, lo que permite usarlo para generar índices, alimentar pruebas o construir flujos deterministas.

range(count): contar desde cero

Cuando recibe un solo argumento, range interpreta ese valor como la cantidad de emisiones. El conteo inicia en 0.

import { range } from 'rxjs';

const src$ = range(5);

console.log('Inicio');
src$.subscribe(value => console.log(value));
console.log('Fin');

El resultado emite 0, 1, 2, 3 y 4. No emite 5, porque el parámetro define la cantidad de valores, no el límite superior inclusivo.

range(start, count): definir inicio y cantidad

Con dos argumentos, el primero define el valor inicial y el segundo la cantidad de emisiones.

import { range } from 'rxjs';

const src$ = range(2, 5);

console.log('Inicio');
src$.subscribe(value => console.log(value));
console.log('Fin');

Este stream emite 2, 3, 4, 5 y 6. El contrato sigue siendo finito y síncrono.

range(start, count, asyncScheduler): diferir la ejecución

asyncScheduler permite programar las emisiones de forma asíncrona. El cambio se observa en el orden de los console.log: primero se ejecuta el código síncrono restante y luego comienzan las emisiones.

import { asyncScheduler, range } from 'rxjs';

const src$ = range(2, 5, asyncScheduler);

console.log('Inicio');
src$.subscribe(value => console.log(value));
console.log('Fin');

interval: emitir valores periódicamente

interval crea un observable asíncrono e infinito. Recibe un periodo en milisegundos y emite un contador incremental en cada tick.

import { interval } from 'rxjs';

const observer = {
  next: (value: number) => console.log('next', value),
  complete: () => console.log('complete'),
};

const interval$ = interval(1000);

console.log('Inicio');
const subscription = interval$.subscribe(observer);
console.log('Fin');

// En una aplicación real, cancela la suscripción según el ciclo de vida.
// subscription.unsubscribe();

interval no emite complete. Si necesitas un número limitado de emisiones, puedes combinarlo con operadores como take, takeUntil o timeout, que analizaremos en artículos posteriores de la serie.

timer: emitir después de un tiempo o iniciar un intervalo diferido

timer también es asíncrono, pero su configuración inicial es distinta. Con un único parámetro numérico, espera esa cantidad de milisegundos, emite 0 y completa.

import { timer } from 'rxjs';

const observer = {
  next: (value: number) => console.log('next', value),
  complete: () => console.log('complete'),
};

const timer$ = timer(2000);

console.log('Inicio');
timer$.subscribe(observer);
console.log('Fin');

Configuraciones especiales de timer

timer acepta configuraciones que cubren varios casos operativos:

• timer(0) agenda la emisión tan pronto como la cola de tareas lo permita, después del trabajo síncrono actual.
• timer(2000, 1000) espera dos segundos y luego se comporta como un intervalo que emite cada segundo.
• timer(fecha) permite agendar la primera emisión para un instante específico usando un objeto Date.

import { timer } from 'rxjs';

const timer$ = timer(2000, 1000);

console.log('Inicio');
timer$.subscribe({
  next: value => console.log('next', value),
  complete: () => console.log('complete'),
});
console.log('Fin');

asyncScheduler: programar trabajo asíncrono

asyncScheduler permite programar tareas usando la infraestructura de schedulers de RxJS. Puede reemplazar casos simples de setTimeout y habilita patrones donde una tarea se reprograma a sí misma.

import { asyncScheduler } from 'rxjs';

const saludar = () => console.log('hola mundo');

console.log('Inicio');
asyncScheduler.schedule(saludar);
console.log('Fin');

También puedes definir un retraso explícito en milisegundos y pasar un único estado a la función programada.

import { asyncScheduler } from 'rxjs';

const saludar = (nombre: string) => console.log(`hola mundo, ${nombre}`);

console.log('Inicio');
asyncScheduler.schedule(saludar, 2000, 'David');
console.log('Fin');

Si necesitas pasar múltiples valores, usa un objeto como estado y desestructúralo en la función. Esto mantiene una firma estable y evita depender de parámetros posicionales adicionales.

Reprogramar una tarea y cancelar la suscripción

Una acción programada puede invocar this.schedule(...) para encadenar la siguiente ejecución. La suscripción resultante permite detener el ciclo.

import { asyncScheduler, SchedulerAction } from 'rxjs';

console.log('Inicio');

const subscription = asyncScheduler.schedule(function (
  this: SchedulerAction<number>,
  state: number
) {
  console.log('state', state);
  this.schedule(state + 1, 1000);
}, 1000, 0);

asyncScheduler.schedule(() => {
  subscription.unsubscribe();
  console.log('Unsubscribe');
}, 6000);

console.log('Fin');

Conclusión

Las funciones de creación definen el contrato inicial del stream. of emite argumentos, from expande estructuras iterables o promesas, fromEvent conecta eventos del DOM, range genera secuencias numéricas, interval modela ticks periódicos y timer cubre emisiones diferidas o intervalos con retardo inicial.

Dominar estas diferencias evita errores sutiles de sincronía, fugas de suscripción y contratos ambiguos entre productores y consumidores. Con esta base, el siguiente paso natural es aplicar operadores para transformar, filtrar y combinar emisiones dentro de una arquitectura reactiva mantenible.