RxJS es la librería JS de referencia para gestionar flujos de datos. OK. Lo has oído varias veces. Pero igual te preguntas… ¿Puedes enseñarme un ejemplo práctico?

De eso justamente va este artículo. Te voy a presentar algunos de los operadores más útiles de RxJS y vas a ver lo potentes que son a través de un ejemplo real: una caja de búsqueda.

Por cierto, si has llegado aquí sin saber bien bien que es RxJS, te recomiendo leer primero mi Introducción a RxJS.

Caja de búsqueda en RxJS

Las cajas de búsqueda son componentes muy habituales en cualquier página web y un caso de uso perfecto para RxJS.

En este ejemplo, simularé un buscador de estados de los EEUU. Cuando realizas una búsqueda, se hace una petición a un «servidor externo» que devuelve los estados que coinciden con la búsqueda. Si el texto de búsqueda está vacío, devuelve todos los estados.

En pocas lineas voy a crear una caja de búsqueda que actualizará los resultados de la vista de forma reactiva, gracias a RxJS. Para hacerlo fácil de entender a todos los niveles, no voy a usar ningún framework, sino simplemente vanilla Javascript. Puedes ver una demo del resultado final en este StackBlitz.

La vista

Como ves en la animación, la vista es muy simple:

• un input de texto
• un botón
• el mensaje de loading…
• y la lista de resultados

El html tiene esta pinta:

<h1>Searchbox rxjs example</h1>

<h3>Seach US States</h3>
<div class="search-box">
  <input id="search-box-input">
  <button id="search-box-btn">Search</button>
</div>

<h3>Results</h3>
<div id="loading" style="display:none">Loading...</div>
<div id="results"></div>

Fíjate en el texto de Loading... Está oculto de entrada, y la idea es mostrarlo mientras se hace la «petición al servidor». Pongo «servidor» porque, para simplificar el ejemplo, el servidor es simulado con una función asíncrona.

El código inicial

Mi código, de momento, solo contiene algunos imports que me van a ser útiles, y selectores para los elementos de la vista:

// mecanismo asíncrono de busqueda de estados
import{ fetchStates } from './states.api';
// utilidades para pintar los resultados
import { appendElementToElement, removeChildrenNodes, populateElementWithList } from './utils';

// Elementos del DOM que quiero manipular
const searchBoxElement = document.getElementById('search-box-input');
const searchBtnElement = document.getElementById('search-box-btn');
const loadingElement = document.getElementById('loading');
const resultsElement = document.getElementById('results');

Empezamos: eventos del input de texto

Lo primero será crear un observable a partir de lo que escribe el usuario sobre el elemento de tipo input.

import  { fromEvent }  from  'rxjs';
import  { map }  from  'rxjs/operators';

// [...resto de imports y elementos del DOM...]

// Search input observable
const searchValue$ = fromEvent(searchBoxElement, 'keyup').pipe(
  map(event => event.target.value),
);

let eventsCount = 0;
// subscription
let eventsCount = 0;
searchValue$.subscribe( search => {
  appendElementToElement(resultsElement, 'DIV', eventsCount + '-' + search);
  eventsCount++;
});

Te lo explico por partes:

• searchValue$: Utilizo la función fromEvent de RxJS, para obtener un Observable a partir de los eventos que emite el elemento searchBoxElement cada vez que se ha levantado una tecla tras presionarla (evento keyup).
• pipe: Es un método de los Observables que me permite encadenarle operadores.
• map: Cada evento contiene mucha información, pero del evento, a mi solo me interesa el value de searchBoxElement. Con el operador map, transformo el evento emitido para transmitir solo el valor del input.
• searchValue$.subcribe: Para que realmente funcione el observable que he creado, necesito suscribirme a éste. Al suscribirme, le paso una función que se ejecutará con cada valor emitido por el observable.

Gracias a map, lo que emite el observable searchValue$ es directamente el texto de la caja de búsqueda. Eso es lo que recibo en la suscripción, y gracias a mi función appendElementToElement, añado cada nuevo evento a mis resultados (resultsElement)

También he creado una variable eventsCount, para que diferencies con claridad cada evento.

El resultado, cuando escribo «York», es este:

Cada vez que escribo una letra, se añade una nueva fila a los resultados.

Un detalle: se ha emitido dos veces la letra Y. Eso es por que ha habido dos eventos de teclado: La tecla y y la tecla SHIFT para hacerla mayúscula.

De hecho, cada vez que toque la letra SHIFT, o cualquier otra tecla que no escriba nada (flechas, Ctrl, etc), se emitirán nuevos eventos.

Operador distinctUntilChanged

Para evitarlo, puedo añadir el operador distinctUntilChanged, que únicamente emite eventos que son diferentes del anterior.

// [...more imports...]
import  { map, distinctUntilChanged }  from  'rxjs/operators';
// [...DOM elements...]

// Search input observable
const searchValue$ = fromEvent(searchBoxElement, 'keyup').pipe(
  map(event => event.target.value),
  distinctUntilChanged(),
);

// [...subscription...] 

Puedes observar la diferencia en la imagen.

Ya no se emiten dos Y seguidas.

De todas formas, no es muy buena idea emitir un evento cada vez que el usuario escribe. Al fin y al cabo, lo que importa es la palabra final que escriba, no cada una de las letras…

Operador DebounceTime

Ahí es donde entra en juego debounceTime. Este operador descarta los eventos que se producen muy seguidos y, cuando se produce una pausa, se limita a emitir el último de ellos.

// [...more imports...]
import  { map, debounceTime, distinctUntilChanged }  from  'rxjs/operators';
// [...DOM elements...]

// Search input observable
const searchValue$ = fromEvent(searchBoxElement, 'keyup').pipe(
  map(event => event.target.value),
  distinctUntilChanged(),
  debounceTime(300),
);

// [...subscription...] 

En este caso, le pido que solo emita eventos que a continuación tienen una pausa de, al menos, 300ms. Dependerá de la velocidad o las ráfagas de escritura del usuario, pero a continuación tienes un ejemplo del resultado esperado al escribir ahora New York:

Botón de búsqueda

Lo que has visto hasta ahora era un calentamiento, para que entendieras mejor cómo funciona RxJS. Ahora lo enlazo con el caso real.

Quiero realizar las búsquedas al hacer click sobre el botón searchBtnElement ¿verdad? ¿Y qué mejor manera que obtener un flujo observable de esos clicks?

Ahí lo tienes, lo llamaré searchClick$.

// [...imports...]
// [...DOM elements...]

// Search input observable
const searchValue$ = /* ...searchValue$ declaration... */

// Search button observable
const searchClick$ = fromEvent(searchBtnElement,  'click');

// [...subscription...] 

Resultados de búsqueda

Ahora viene lo bueno: Combino cada evento de searchClick$ con el último valor emitido por searchValue$ para realizar la búsqueda. Este flujo lo llamaré search$.

// [...more imports...]
import  { map, debounceTime, distinctUntilChanged, withLatestFrom, switchMap }  from  'rxjs/operators';
// [...DOM elements...]
// [...searchValue$ and searchClick$ declarations...]

// search composed observable
const search$ = searchClick$.pipe(
  withLatestFrom(searchValue$, (click, search) => search ),
  distinctUntilChanged(),
  switchMap( search => fetchStates(search)),
);

// [...subscription...] 

De nuevo, te lo explico por pasos:

• withLatestFrom: Este operador se suscribe a searchValue$, de modo que cuando searchClick$ emite un click, el operador lo combina con el último valor emitido por searchValue$. En este caso, además, estoy usando la función de proyección (es opcional), para que en vez de devolverme ambos datos, me devuelva solo el que me interesa (la búsqueda).
• distinctUntilChanged: Ya lo has visto antes. En este caso, quiero evitar búsquedas consecutivas con el mismo valor de búsqueda (imagina varios clicks del usuario, sin que el texto cambie).
• switchMap: Este operador reemplaza el flujo actual de datos por el de un nuevo Observable (al que se suscribe internamente). En este caso, al recibir un evento, devuelve mi función fetchStates, pasándole el texto de búsqueda recibido.

 

Mi función fetchStates simula la búsqueda al servidor y devuelve un Observable que emite los resultados de búsqueda. Así que switchMap reemplaza mi flujo por los eventos de este nuevo observable.

Con este código, he pasado de tener un flujo de textos de búsqueda, a tener un flujo de resultados del servidor, dado un cierto texto.

La suscripción

Ahora que ya tengo los resultados del servidor, puedo eliminar la suscripción que tenía antes y suscribirme, en cambio, al Observable search$.

// [...more imports...]
import  { map, debounceTime, distinctUntilChanged, withLatestFrom, switchMap }  from  'rxjs/operators';
// [...DOM elements...]
// [...searchValue$ and searchClick$ declarations...]

// search composed observable
const search$ = searchClick$.pipe(
  withLatestFrom(searchValue$, (click, search) => search ),
  distinctUntilChanged(),
  switchMap( search => fetchStates(search)),
);

// search subscription
search$.subscribe( data => {
  removeChildrenNodes(resultsElement);
  populateElementWithList(resultsElement, data);
});

Ahora, cuando hago click sobre el botón de búsqueda, se realiza una petición al «servidor» y recibo el resultado en la suscripción. Allí, me encargo de renderizarlo gracias a mis funciones removeChildrenNodes y populateElementsWithList.

El resultado, sería este:

Petición inicial: operador startWith

Esto ya empieza a tener buena pinta. El único problema es que, de entrada, mis lista de estados aparece vacía.

Ya que cuando le paso un string vacío a fetchStates, éste me devuelve todos los estados, lo ideal sería lanzar una búsqueda vacía al inicio de la suscripción. Todo controlado, para eso esta el operador startWith.

// [...more imports...]
import  { map, debounceTime, distinctUntilChanged, withLatestFrom, switchMap }  from  'rxjs/operators';
// [...DOM elements...]

// [...searchValue$ and searchClick$ declarations...]

// search composed observable
const search$ = searchClick$.pipe(
  withLatestFrom(searchValue$, (click, search) => search ),
  startWith(''), //emit initial empty string 'search' value
  distinctUntilChanged(),
  switchMap( search => fetchStates(search)),
);

// [search subscription]

Como ves, uso startWith('') para forzar una búsqueda inicial con un string vacío.

Nota: El orden de los operadores importa. El lugar en el que he colocado el startWith no es casual.
* Si lo hiciera antes del withLatestFrom, no tendría ningún resultado, porque el flujo se quedaría bloqueado ya que searchValue$ aún no ha emitido ningún valor.
* Si lo hiciera después del switchMap, no se realizaría la petición al servidor: En su lugar, search$ simplemente emitiría un string vacío.

Mecanismo de «loading»

Cuando tienes una caja de búsqueda, es recomendable dar algo de feedback al usuario.

En este caso, los resultados no se muestran de inmediato porque fetchStates simula una cierta latencia. Por eso, desde que se produce la petición, hasta que se obtienen los datos, voy a mostrar el mensaje Loading… en el lugar donde deberían ir los resultados. Y lo voy a hacer, de nuevo, con RxJS.

Los eventos de loading

La vista HTML que te he mostrado al principio tenía un DIV con el texto login y, en el código inicial, había una línea para acceder a este elemento:

const loadingElement = document.getElementById('loading');

Para manipularlo de forma reactiva, lo ideal sería disponer de un flujo observable que emita valores true o false en función de si hay que mostrar o no este elemento.

Así:

import  { fromEvent,  Observable  }  from  'rxjs';
// [...more imports...]
// [...DOM elements...]
// [...searchValue$, searchClick$ and search$ declarations...]

// loading state observable
const loading$ = new Observable();

// loading event subscription
loading$.subscribe(isLoading => {
  loadingElement.style.display = isLoading ? 'block' : 'none';
  resultsElement.style.display = isLoading ? 'none' : 'block';
});

// [search subscription]

Como ves en la suscripción, cuando loading$ emite un evento true, muestro loadingElement, y oculto resultsElement. Al recibir un evento false, hago lo contrario. Escondo el loading y muestro los resultados.

Pero claro… ¿como emito esos valores true/ false?

Emitiendo eventos mediante Subjects

Para hacer que un observable emita eventos de forma arbitraria, lo habitual es recurrir a un Subject. El Subject de RxJS es una clase especial de Observable, que además es un Observer.

El Subject es una especie de HUB: Puede recibir eventos con el método next (como un Observer) y a su vez los distribuye a sus suscriptores, como un Observable.

Por claridad, suelen separarse ambos lados. Fíjate:

import  { fromEvent,  Subject  }  from  'rxjs';
// [...more imports...]
// [...DOM elements...]
// [...searchValue$, searchClick$ and search$ declarations...]

// loading state observable
const loadingSubject =  new  Subject<boolean>();
const loading$ = loadingSubject.asObservable().pipe(startWith(false));

// [loading event subscription]
// [search subscription]

Por un lado tengo loadingSubject, que es el que usaré para decidir qué valores emitir. Por el otro, tengo loading$, que es la parte observable del primero y es donde me voy a suscribir.

Ahora solo me falta emitir esos trueo false, pero… ¿cuando?

Cuando realmente sé si estoy realizando la búsqueda o he terminado, es dentro del flujo search$. Ahí es donde puedo usar el operador tap (que sirve para ejecutar acciones colaterales) para emitir valores gracias al método next del Subject.

Veamos:

import  { tap, /*...other operators...*/ }  from  'rxjs/operators';
// [...more imports...]
// [...DOM elements...]
// [...searchValue$ and searchClick$ declarations...]

// search composed observable
const search$ = searchClick$.pipe(
  withLatestFrom(searchValue$, (click, search) => search ),
  startWith(''),
  distinctUntilChanged(),
  tap(() => loadingSubject.next(true)),
  switchMap( search => fetchStates(search)),
  tap(() => loadingSubject.next(false)),
);

// loading state observable
const loadingSubject =  new  Subject<boolean>();
const loading$ = loadingSubject.asObservable().pipe(startWith(false));

// loading event subscription
loading$.subscribe(isLoading => {
  loadingElement.style.display = isLoading ? 'block' : 'none';
  resultsElement.style.display = isLoading ? 'none' : 'block';
});

// [search subscription]

Como ves, antes de hacer la petición al servidor, le paso el valor true al Subject, que a su vez, lo emitirá a través del observable loading$ (actualizando la vista).

Al completarse la petición (cuando switchMap emite el evento de fetchStates), uso de nuevo el Subject para emitir el valor false a través de loading$, escondiendo el texto «loading…» y mostrando los resultados de la búsqueda.

A continuación puedes ver el proyecto StackBlitz con el resultado final.

Operadores que has visto

Como ya sabrás a estas alturas, los operadores de RxJS son funciones puras con una aproximación funcional, que puedes aplicar de forma encadenada sobre un flujo de datos.

Te hago un breve resumen de los operadores que has visto:

• map: Te permite transformar el evento de entrada. Tiene esta estructura map(data => transform(data)).

• startWith: Te permite añadir un evento o secuencia de eventos al inicio de flujo de datos. Por ejemplo, si quieres que tu flujo de datos, sea el que sea, empiece con el valor cero, podrías usar: startWith(0).

• debounceTime: Descarta eventos que pasan de forma muy frecuente y solo deja pasar aquellos tras los que hay un cierto tiempo de guarda. Ideal para evitar varios clicks seguidos, por ejemplo. debounceTime(timeInMs)

• distinctUntilChanged: Solo deja pasar un evento si es distinto del anterior evento transmitido. distinctUntilChanged()

• tap: No es tanto un efecto pensado para alterar el flujo de datos, sino para facilitar efectos colaterales. Por ejemplo, si quieres guardar cada evento en localstorage, podrías hacer: tap(event => localStorage.setItem('evt', event)).

• switchMap: Forma parte de los denominados High Order Observables. En este caso, reemplaza el flujo actual por los eventos emitidos por un nuevo observable. La estructura sería esta switchMap(event => newObservableBasedOnEvent(event)).

• withLatestFrom: Combina el evento actual del flujo, con el evento más reciente emitido por otro observable. Devuelve un array, con ambos eventos. La estructura sería esta withLatestFrom(anotherObservable) y a su salida, emite un array [originalObservableEvent, anotherObservableEvent]. En el ejemplo, además, he usado una función de proyección para transformar el formato de salida.

Resumen

Con este cuarto artículo, concluye la saga Aprende RxJS desde cero.
Espero que te hayan servido para entender mejor como funciona esta fantástica librería y atreverte con ella.

Si te ha sabido a poco y quieres conocer más operadores, ver más ejemplos, o trabajar con casos más complejos, te recomiendo 100% mi curso RxJS Nivel PRO.