Reactivex - rxjs: de cero hasta los detalles download

Este es el segundo artículo en mi serie de programación reactiva en JavaScript. Si no tienes idea de qué es programación reactiva, puedes pasar por el primer artículo: Programación Reactiva en JavaScript.

Table of Contents

• Introducción
• Creando Observables
• Observable Creators
• Pipeable Operators
• Obteniendo data a partir de dos o más Observables
• High Order Observables
• Hot ‘n Cold

Muchos de los conceptos que tratamos aquí están explicados en el artículo anterior.

Si ya te sientes cómodo con los conceptos tratados en este y ese artículo, te recomiendo pasar a leer el tercer artículo: Programación reactiva e iterable con Callbags, que trata de una especificación para trabajar con streams de datos iterables (pull based) y reactivos (push based) a partir de funciones.

Igualmente he publicado un video haciendo una implementación de un Observable desde cero.

Si has desarrollado aplicaciones con Angular (2+) me imagino que en algún momento has tenido que lidiar con RxJS.

Recuerdo la primera vez que lo hice.

No fue lindo…

Sentí genuinamente que estaba frente a algo fuera de mis capacidades. Algo muy complejo de comprender. Así que me hice de ánimos, abrí google, stackoverflow y un par de blogs que encontré en el camino y me tiré de cabeza esperando entender cómo funcionaba esta librería.

Un par de meses después, viendo una charla en youtube (o leyendo un artículo, ya no recuerdo) tuve ese momento de claridad y todo me hizo sentido. Después de mucho esfuerzo y de darme incontables cabezazos contra el teclado (y de escribir mucho código que no entendí del todo como funcionaba), logré comprender el paradigma base de la programación reactiva .

Desde entonces ha sido una bonita relación.

En fin, no fue hasta hace poco que caí en la cuenta de que una de las mayores dificultades al aprender RxJS no tiene que ver con entender la librería en sí, sino con el paradigma sobre el cual sienta sus bases.

Espero que tras haber leído este artículo, tengas el conocimiento técnico base para resolver cualquier problema que se te presente con RxJS.

Introducción

RxJS is una librería de programación reactiva cuyo fin es simplificar la composición de código asíncrono y basado en eventos a través de secuencias observables.

RxJS provee una estructura de datos llamada Observable, estructuras derivadas como Observer, Scheduler, Subject y operadores para manipular estas estructuras, inspirados en los métodos que podemos encontrar en Array.prototype.

RxJS combina patrones de diseño de software como el Patrón del Observador, el Patrón del Iterador y conceptos de programación funcional utilizando colecciones para modelar una forma ideal de manejar secuencias de eventos.

Los tres grandes focos de dificultad al aprender RxJS son los siguientes:

1. La librería acuña muchos conceptos con los que no estamos tan familiarizados al hacer programación proactiva.
2. La programación reactiva se ocupa de modelar relaciones entre eventos a través del tiempo. Cada vez que involucramos la variable tiempo en nuestro código implica incremento en complejidad.
3. RxJS cuenta con una gran cantidad de operadores y funciones, por lo que puede parecer abrumadora.

Pues bien, la primitiva base de RxJS es el observable. En su estado más esencial, un observable se adhiere a la siguiente interfaz:

Un observable es un objeto con un método subscribe que toma como parámetro un observador y nos retorna una suscripción. El observador se suscribe al observable y cada vez que el observable emite un valor, el observador es notificado. La suscripción que nos retorna el método subscribe, posee un método llamado unsubscribe que podemos llamar cuando deseamos terminar la relación entre observable y observador.

Si esto te resulta confuso, podemos pensar en varias situaciones en la vida real en la que nos suscribimos a una fuente de eventos.

Por ejemplo, cuando vamos a la fiambrería o a la farmacia y sacamos un número de atención, nos estamos suscribiendo a un stream de números de atención. Cada vez que un número nuevo se despliega en el tablero electrónico, estamos pendientes de si el número emitido corresponde a nuestro número de atención (digamos que sufrimos de mala memoria y tenemos que chequear constantemente cuál es nuestro número).

Cuando llega nuestro turno, somos atendidos y dejamos de estar pendientes de qué números vienen a continuación. En cierta forma, cancelamos la suscripción que teníamos inicialmente con el stream de números de atención.

Volveremos a revisar el caso del Turnomatic 8000 en más detalle hacia el final del artículo para entender el concepto de multicast.

Creando Observables

Pues bien, dicho sea lo anterior ¿Cómo produce un observable los valores que va a emitir?

RxJS nos provee una clase genérica llamada Observable que podemos utilizar para generar objetos que se adhieren a esta especificación. Queda de nuestro lado definir qué valores van a emitir nuestros observables.

Digamos que queremos crear un observable a partir de algún evento en el DOM. Para esto crearemos una función llamada fromEvent:

De igual forma podemos definir nuestro observable usando la función create:

La función que reciben create y Observable como parámetro es la función de producción. Esta función tiene como parámetro un objeto (al que llamamos observer) con los métodos next, error y complete.

Para nuestra implementación de fromEvent no tiene mucho caso usar las funciones error y complete, las veremos más adelante en su momento.

Dentro de la función de producción registramos un event listener utilizando el nodo y el nombre del evento que le pasaremos como argumento a fromEvent. La función de producción retorna otra función que se encargará de remover el event listener del nodo al momento de terminar la suscripción.

Utilizamos fromEvent de la siguiente forma:

Nuestra función fromEvent retorna una nueva instancia de Observable, a la que hemos llamado convenientemente click$ .

Los observables al igual que las funciones son flojos por naturaleza. ¿A qué se refiere esto? Pues bien, una función no hace absolutamente nada hasta que es invocada. De igual forma un observable no hace absolutamente nada hasta que llamamos al método subscribe.

Esta es la primera regla que debemos recordar al usar RxJS:

1. Siempre debemos suscribirnos a nuestros observables si queremos que hagan algo.

Esto significa que al llamar a la función fromEvent:

Aún no ha pasado nada en nuestro programa. El event listener no se ha registrado en el elemento $button ni nada por el estilo. La función que hemos pasado al instanciar la clase Observable no se ejecutará hasta que llamemos al método subscribe:

Cada vez que llamamos al método next del observador dentro del callback que le hemos pasado a Observable, estamos produciendo los valores que nuestro observable emitirá. A esto se le denomina productor o producer y es vital tenerlo en consideración al momento de trabajar con observables. Más adelante veremos por qué.

Genial, hemos creado nuestro primer observable ¡hurrah!.

Observable Creators

Afortunadamente no son muchas las ocasiones en que crearemos observables de forma manual (me atrevería a decir que NUNCA deberíamos crear observables manualmente), pues RxJS nos provee de múltiples operadores para simplificar este proceso. A estos operadores se les conoce como Observable Creators y los puedes importar directamente desde rxjs. Nuestro ejemplo anterior quedaría de la siguiente forma:

¿Te has dado cuenta que he usado como sufijo el símbolo $ en la constante click$?

Pues esto no es más que una convención para nombrar a los observables. Como nuestro observable representa una colección de clicks a través del tiempo, reemplazamos la s en clicks por un signo $.

Dependiendo del tipo de estructura de dato que queremos observar (y de lo que queremos hacer con ella) podemos usar distintos Observable Creators.

Pues bien, digamos que necesitamos crear un stream a partir de uno o más valores. Para esto podemos ocupar el operador of.

of crea una secuencia observable a partir de los valores que se le pasen por argumentos. Al emitir todos los valores luego se completa.

Una vez que se han emitido todos los valores al ejecutar el callback correspondiente a next, luego se ejecuta el tercer callback correspondiente a complete.

Otro ejemplo:

Cuando definimos fromEvent al comienzo del artículo, mencioné que no tenía sentido implementar error y complete en ese caso puntual y es porque básicamente nunca sabremos cuando el usuario de nuestra aplicación va a dejar de hacer click en el botón.

En el caso de of, la cantidad de emisiones que tendrá el observable sí está restringida por el número de argumentos que le pasamos, por lo mismo tiene sentido que el observable se complete a diferencia de fromEvent.

Esta es la segunda regla que debemos recordar al usar RxJS:

2. Hay Observables que se completan y otros que no.

of es bastante útil para crear observables de casi cualquier cosa, pero existe un operador aún más flexible y es from.

from crea una secuencia observable a partir un arreglo, un objeto pseudo-arreglo, una promesa, un iterable o un objeto pseudo-observable.

from es probablemente uno de los operadores más útiles, ya que nos permite tomar casi cualquier estructura de datos y convertirla en una secuencia observable.

A diferencia de of, from aplana la estructura que le pasamos, emitiendo un valor por cada elemento de esta.

Intentemos crear un observable a partir de una estructura que implementa [Symbol.iterator], tal como URLSearchParams:

Veámoslo ahora con promesas:

Considerando esto podríamos crear un pequeño contenedor para envolver todas las llamadas http usando fetch en un observable, por ejemplo:

Afortunadamente ya no tienes que crear tu propio wrapper de fetch, ya que desde la versión 6.5.0 puedes importar el operador fromFetch desde 'rxjs/fetch'

Pipeable Operators

Cuando trabajamos con colecciones de datos en nuestros programas pocas veces nos basta con solamente tener los datos en sí. Probablemente debemos procesarlos, filtrarlos u obtener datos derivados a partir de estos.

Sería agradable poder razonar sobre colecciones de datos asíncronas y eventos de la misma manera que lo hacemos con el resto de nuestra data. RxJS nos permite manipular, procesar, y convertir estos eventos en algo que podamos usar en nuestros programas de forma secuencial.

Para esto RxJS nos proporciona los pipeable operators, que son funciones que nos permiten transformar la data que nuestros observables emiten. Estos operadores se pasan como parámetro al método pipe de la siguiente forma:

Desde la versión 7 de RxJS todos los operadores están expuestos en 'rxjs'. Si estás trabajando con la versión 6, estos se encuentra en 'rxjs/operators'

En este caso lo que hacemos es mapear cada click en el documento a un objeto con las coordenadas x e y del click.

Podemos aplicar todos los operadores que queramos en un pipeline:

Utilizamos el observable creator interval que toma como parámetro un número n de milisegundos y emite un número (partiendo desde 0) cada n milisegundos.

Filtramos cada uno de los número emitidos por interval por medio de filter, que recibe como parámetro una función que debe retornar true o false. Si el valor retornado por filter es true, el valor emitido pasa al siguiente operador en la cadena de operadores, de lo contrario, el número que recibe filter como argumento nunca será pasado al siguiente operador del pipeline, en este caso, a scan.

scan, que es muy parecido a Array.prototype.reduce, toma el último valor emitido por el Observable y el valor actual y retorna la suma de ambos valores.

map recibe la suma producida por scan y retorna un string basado en ese valor.

Lo importante es tener en cuenta que cada valor emitido por el Observable inicial, será procesado de forma independiente por el pipeline de operadores. Nuevamente cabe mencionar que ninguna de estas operaciones se realizará, si es que no nos suscribimos al Observable.

Podemos representar el ejercicio anterior con el siguiente marble diagram o diagrama de canicas:

Obteniendo data a partir de dos o más Observables

En muchos casos debemos combinar dos o más fuentes de eventos para poder realizar alguna acción en nuestra aplicación.

Digamos que tenemos un reporte o métrica que debemos mostrar con información consultada de una API cada 1 minuto. De igual forma, tenemos un botón que podemos presionar para requerir el reporte enseguida.

Tenemos dos fuentes de eventos que podrían gatillar un llamado a la API para obtener nuestro reporte. Podríamos hacer algo así:

Que es una alternativa válida, pero existe una mejor implementación si combinamos ambos observables. Para efectos de este ejemplo, sólo nos importa cuando cualquiera de los observables emita un valor, por lo que podemos mergear ambos observables y suscribirnos esta nueva fuente de eventos:

En este caso puntual no nos importa el valor emitido por click$ y everyMinute$. Sólo nos importa saber cuándo ambos observables emiten valores. En otras ocasiones sí nos puede importar qué valores son emitidos por los observables que estamos componiendo.

Un caso distinto sería si el reporte dependiera de un grupo de filtros seleccionados por el usuario. Ahora nos interesa saber el último valor de cada filtro para disparar la petición a la API. En este caso podemos usar combineLatest para agrupar nuestros observables:

En general podemos usar distintas estrategias para combinar observables dependiendo de qué tipo de efecto queremos disparar a partir de ellos.

High Order Observables

Si el concepto de higher order functions te resulta familiar, pues buenas noticias, los higher order observables son lo mismo, pero con observables. (Si no sabes qué es una función de orden mayor, escribí un artículo que te puede ser de ayuda).

En pocas palabras, un higher order observable, es un observable que emite observables como valores.

Valiéndonos del último ejemplo, hasta el momento no hemos definido la implementación de fetchReport. Digamos que lo único que hace esta función es hacer una petición GET a un API pasando como query params los valores indicados en los filtros. Si quisiéramos utilizar el wrapper que hicimos de fetch para hacer esa consulta nuestro código quedaría algo así:

Si bien esta es implementación funciona, nos estamos suscribiendo a http desde dentro del callback de suscripción a filter$. Esto es claramente un antipatrón, ya que si nos suscribimos a otro observable dentro del callback de suscripción a http, terminamos básicamente con otra encarnación del difamado callback hell:

En nuestro caso podríamos mapear los eventos provenientes de filter$ directamente a fetch:

Como http retorna un observable, lo que obtenemos en res$ es básicamente la respuesta de la llamada al endpoint como observable, por lo que nuestro problema aún no se ha resuelto (pues tendremos que suscribirnos a res$ dentro del subscribe). Para solucionar esto podemos usar un operador como switchMap:

Lo que hace por debajo switchMap es suscribirse internamente por nosotros al observable que estamos retornando en el operador y nos retorna la data emitida por ese observable. Igualmente cada vez que filter$ emite un nuevo valor, switchMap cancela la suscripción anterior e inicia una suscripción nueva.

Hot ‘n Cold

Los observables al igual que Alexander Rodríguez pueden ser fríos o calientes.

Un observable es frío cuando el estado en la función de producción no se comparte entre sus suscriptores; y es caliente cuando sí es compartido. El concepto de frío/caliente también se conoce unicast/multicast.

¿Recuerdas que al comienzo del artículo hablamos del Turnomatic 8000 (bañado en súper oro, con incrustaciones de diamantes y engranajes de plastigoma)?

El Turnomatic 8000 es un ejemplo de multicast, pues el estado del observable se comparte entre todos sus suscriptores.

Cuando llegas a la fiambrería o farmacia y tomas un número, el número de atención en turno emitido por el Turnomatic 8000 es igual para todos. Todas las personas que están esperando ser atendidas ven el mismo número, por lo que podemos concluir que estado del Turnomatic 8000 se comparte entre todos sus suscriptores.

Los observables en RxJS son mayoritariamente unicast. Esto implica que un observable va a emitir un set “fresco” de valores para cada uno de sus suscriptores:

En este caso cada nuevo suscriptor a clock$ parte desde 0, ya que la función de producción se ejecuta cada vez que nos suscribimos al observable.

Si quisiéramos compartir el estado entre todos los suscriptores de un observable podemos usar operadores como share, entre otros:

Con todo lo que hemos aprendido hasta ahora, podemos establecer la tercera regla que debes recordar al usar RxJS:

3. Un stream es frío cuando la función de producción se ejecuta para cada nuevo suscriptor; y es caliente cuando la función de producción se ejecuta sólo una vez (con la primera suscripción).

Subjects

Para terminar (¡al fin!), hablaremos a grandes rasgos de Subject, que es una abstracción muy similar al Observable, ya que nos podemos suscribir a ellos. La gran diferencia entre Subject y Observable, es que el primero es usado principalmente para multicast, mientras que el segundo es generalmente unicast como ya sabemos.

Igualmente Subject tiene los métodos next, error y complete por lo que puede ser usado como un observador para convertir un observable frío a uno caliente:

En este ejemplo lo que hacemos es que en vez de suscribirnos directamente a clock$— que correría la función de producción de interval para cada nueva suscripción — suscribimos nuestro Subject como intermediario y luego nos suscribimos a sub, de esta forma nos aseguramos que la función de producción se ejecute sólo una vez (ya que existe solo una suscripción a clock$) y que todos los suscriptores de sub reciban los mismos valores.

Existen varios tipos de Subjects que sirven para distintos casos, pero eso es material para otro artículo.

Si has llegado hasta aquí, felicitaciones y muchas gracias. Me ha costado mucho completar este artículo y estoy muy feliz de que lo hayas leído y de que, afortunadamente, te haya sido de utilidad.

Si te ha gustado el contenido no olvides dejar una par de claps y seguirme en Twitter que tengo Twitter y seguirme por acá en Medium igualmente.