RxJs – Guía para principiantes

¿Alguna vez te has preguntado cómo puedes abrir un archivo de 10 GB en un dispositivo con solo 4 GB de RAM? ¿O cómo la aplicación Netflix puede reproducir una película 4K alojada en la nube en su teléfono tan pronto como presione reproducir? En un dispositivo con 4 GB de RAM, un archivo de 10 GB es efectivamente infinito. ¿Cómo se las arregla tu dispositivo para cargar algo que es infinito?

Tiene que cargar el archivo en la memoria en pequeños fragmentos, leerlos y descartarlos antes de cargar más datos en la memoria. Tiene que transmitir los datos y procesarlos en pequeños fragmentos.

¿Qué es una corriente?

La secuencia es una colección de datos que es potencialmente infinita. Es una secuencia de datos que llegan en horas extras. Se puede considerar como elementos en una cinta transportadora que se procesan uno a la vez.

Stream = Array + Infinity

Dado que los datos son potencialmente infinitos, nuestros bucles de confianza no serán efectivos en ellos. No puede escribir un bucle for de cero a infinito para procesar todo el flujo.

for (let i = 0; i < infinite; i++) {
  const element = stream[i];
}

El problema es cómo sabemos cuándo detenerlo. Aquí es donde los Observables entran en escena.

observables

Los observables son colecciones potencialmente infinitas que devuelven valores uno a la vez de forma asíncrona. es decir, potencialmente puede pasar algún tiempo entre un valor devuelto y el siguiente.

Observable = Array + Infinity + Asynchronous
// OR
Observable = Promise + Returns many times

————— Value 1 ————— Value 2 ————— Value 3 ————— Value 4 —————|—>

En este sentido, son muy similares a Promises . Las promesas pueden devolver un valor después de que haya pasado algún tiempo. Observables devuelve valores potencialmente infinitos con un tiempo que pasa entre cada uno de ellos.

// Two states => resolve, reject
const promise = new Promise(resolve, reject);
  
promise
.then((data) => console.log("Data came back:" + data)) // Success
.catch((err) => console.error("No, Ew David", err)); // Error

Las promesas tienen dos estados posibles: resolver, rechazar, o en otras palabras: completar, error.

const observable = from([1, 2, 3, 4]);
  
// Three states => next, complete, error
observable.subscribe({
  next: (value) => console.log("Next value:", value),
  complete: () => console.log("Infinity is Done!!! ¯\_(ツ)_/¯ "),
  error: (err) => console.log("No, Ew Binod", err),
});

Los observables agregan un estado adicional al mismo concepto: siguiente, completo, error. Una de las bibliotecas Observable más populares en JavaScript es RxJS . Lo que hace que RxJS sea increíble no es solo el concepto de Observables, sino también la amplia gama de Operadores . Estos operadores pueden actuar en los observables para permitir que el código asíncrono complejo se componga fácilmente de manera declarativa.

Operadores RxJs

En RxJS, los operadores son funciones que aceptan un Observable como entrada, ejecutan algunas transformaciones en él y devuelven el nuevo Observable transformado como salida. Estos operadores son (en su mayoría) funciones puras y sin efectos secundarios; es decir, no cambian el Observable entrante de ninguna manera. Esto hace que los operadores se puedan enstringr o canalizar; permitiendo que la lógica asíncrona en su aplicación se divida en pequeños bloques manejables.

RxJs clasifica a sus operadores en algunas categorías, pero los operadores más utilizados son los operadores de creación y los operadores de transformación . En esta guía, exploraremos cómo crear un Observable, cómo transformar y cómo consumir los datos emitidos por los Observables.

Operador de creación: de

Esta es la forma más sencilla de crear un Observable a partir de datos estáticos. Es un contenedor fácil de usar que acepta cualquier secuencia de datos como entrada y devuelve un Observable listo para usar. Este operador es útil para iniciar una nueva canalización de Observable.

of(10, 20, 30).subscribe(
  next => console.log('next:', next),
  err => console.log('error:', err),
  () => console.log('the end'),
);

Producción:

'next: 10'
'next: 20'
'next: 30'
the end

Operador de creación: desde

Este Operador es similar a `of` pero funciona para datos iterables, es decir, acepta una colección de datos y devuelve un Observable que emite cada valor de la colección uno a la vez. El poder real de este Operador proviene del hecho de que también puede aceptar iterables asincrónicos como generadores, promesas y otros Observables.

from([10, 20, 30]).subscribe(
  next => console.log('next:', next),
  err => console.log('error:', err),
  () => console.log('the end'),
);
  
console.log('----------')
  
const promise = fetchDataFromServer();
from(promise).subscribe(
  next => console.log('next:', next),
  err => console.log('error:', err),
  () => console.log('the end'),
);

Producción:

'next: 10'
'next: 20'
'next: 30'
the end
----------
'next: {msg: "Hello world!"}'
the end

Operador de transformación: mapa

Este operador es muy similar a Array#map. Acepta cada nuevo valor emitido por el observable, lo transforma y lo pasa al siguiente Operador en la tubería. Aquí es donde la base conceptual de Streams and Observables comienza a brillar.

¿Por qué tomarse la molestia de aprender todo este nuevo concepto cuando el mismo problema se puede resolver usando Array#map ? Los observables son muy útiles cuando simplemente no podemos cargar todo el conjunto de datos en una array (es decir, los datos son efectivamente infinitos). O cuando no tenemos todo el conjunto de datos disponible por adelantado. Es decir, el conjunto de datos es asíncrono y los nuevos valores ingresan lentamente a través de la red. O muchas veces tenemos ambos problemas, lo que significa que, de hecho, ingresan datos infinitos lentamente a través de la red, unos pocos valores a la vez. 

// Another RxJS creation operator that
// starts at 0 and emits 1000 values
range(1, 1000) 
.pipe(map(x => x * 10))
.subscribe(
  next => console.log('next:', next),
  err => console.log('error:', err),
  () => console.log('the end'),
);

Producción:

'next: 10'
'next: 20'
'next: 30'
....
....
....
'next: 10000'
the end

Los operadores RxJs casi siempre son puros/sin efectos secundarios y funcionan con un valor emitido a la vez. Esto hace que manejar conjuntos de datos efectivamente infinitos sea realmente fácil. Dado que la función no tiene efectos secundarios, el sistema no tiene que retener elementos que no se están procesando actualmente. es decir, sólo un elemento a la vez se mantiene en la memoria.

Operador de transformación: mergeMap

Este operador es muy similar a map pero su función de transformación devuelve datos asíncronos (Observables o Promises). Esto hace que el manejo de muchas llamadas asíncronas a servidores o bases de datos sea muy fácil e incluso nos permite paralelizar esas llamadas.

range(1, 1000).pipe(mergeMap(pageNum => 
    fetchBulkDataFromServer({pageNum: pageNum})))
.pipe(map(bulkData=>`Page Num ${bulkData.page} 
    returned ${bulkData.items.length} items`))
.subscribe(
    next => console.log('next:', next),
    err => console.log('error:', err),
    () => console.log('the end'),
);

Producción:

'next: Page Num 1 returned 100 items'
'next: Page Num 2 returned 90 items'
'next: Page Num 3 returned 70 items'
'next: Page Num 4 returned 100 items'
....
....
'next: Page Num 1000 returned 30 items'
the end

Como mergeMap está mapeando sobre datos asíncronos (Observables), acelera significativamente las cosas al mapear sobre múltiples Observables en paralelo. Acepta un segundo argumento `recuento de concurrencia` que define cuántos Observables ejecutar en paralelo. Implementar este nivel de procesamiento asincrónico paralelo sin usar Observables no es una tarea sencilla y puede resultar fácilmente en errores de concurrencia difíciles de depurar.

const maxParallelApiCalls = 50;
range(1, 1000).pipe(mergeMap(pageNum =>
    fetchBulkDataFromServer({pageNum: pageNum}),
maxParallelApiCalls)).pipe(map(bulkData =>
    `Page Num ${bulkData.page} returned 
    ${bulkData.items.length} items`))
.subscribe(
    next => console.log('next:', next),
    err => console.log('error:', err),
    () => console.log('the end'),
);

Producción:

'next: Page Num 7 returned 10 items'
'next: Page Num 12 returned 8 items'
'next: Page Num 38 returned 12 items'
'next: Page Num 3 returned 70 items'
....
....
'next: Page Num 1000 returned 30 items'
the end

En el ejemplo anterior, RxJs comienza a procesar 50 Observables al mismo tiempo y emite los datos devueltos por esos Observables en el orden en que terminaron. Entonces, cualquiera que sea la llamada API que devuelva primero, sus datos se canalizarán al siguiente operador. Aquí hay una visualización de línea de tiempo de cómo mergeMap paraleliza los datos asincrónicos .

<Start> — Value 1 —————————————————————————————————|—>
<Start> ————————————————————— Value 2 —————————————|—>
<Start> ——————————— Value 3 ———————————————————————|—>
<Start> —————————————————————————————— Value 4 ————|—>
<Start> —————————————————Value 5 ——————————————————|—>
——————————————————————— Merge ————————————————————————
<Start> — Value 1 —— Value 3 —— Value 5 —— Value 2 —— Value 4 —|—>

Conclusión: los ejemplos anteriores cubren algunos operadores en esta guía para principiantes, pero RxJS tiene muchos más, adecuados para una amplia variedad de casos de uso. Consulte su documentación para explorar más.

Referencia: https://rxjs.dev/api