🤺 🔈 🏇🏻 Estudiamos operadores RxJS multicast 🦒 🙆🏽 🚊

Hola Habr! Les presento la traducción del artículo "Comprender los operadores de multidifusión RxJS" de Netanel Basal.

Los operadores de transmisión o multidifusión a menudo parecen ser el tema más difícil de aprender sobre RxJS. En este artículo intentaré explicar todo de manera accesible.

Consideraremos la estructura interna de los operadores de multidifusión y las tareas que resuelven.

Comencemos describiendo los componentes básicos de RxJS.

Observable

En RxJS, los objetos observables (en adelante denominados "flujos") están inicialmente fríos. Esto significa que cada vez que se suscribe a una transmisión, se realiza una devolución de llamada de la suscripción.

Para una mejor comprensión, cree la siguiente implementación:

class Observable {
  constructor(subscriptionFn) {
    this.subscriptionFn = subscriptionFn;
  }

  subscribe(observer) {
    return this.subscriptionFn(observer);
  }
}

El constructor Observableacepta un solo parámetro: la devolución de llamada de la suscripción
subscriptionFn. Se llamará cada vez que nos suscribamos a stream ( subscribe()).

A veces también llaman a una devolución de llamada de una suscripción producer, ya que también "produce" valores para el suscriptor (objeto observador en nuestro código).

El método subscribe()toma una entrada observer. Es un objeto con tres propios métodos: next(), error(), complete(). En Live RxJS, puede pasar tres funciones en lugar de un objeto.

El método, subscribe()cuando se llama, llama a la función de suscripción y lo pasa a la entrada observer.

No mencionamos el método ahoraunsubscribe, pero debe tenerse en cuenta que cada suscripción proporciona una forma de destruirla. Muy a menudo, una suscripción devuelve una función (o un objeto con un método apropiado), durante el cual se destruye la conexión entre la secuencia y sus suscriptores.

Todo esto es bastante simple. Acerquémonos más a la realidad ahora. Por ejemplo, envuelva una API XHR nativa en una secuencia


function http(url) {
  // This function will be called when we call http().subscribe()
  const subscriptionFn = observer => {
    log('Observable execution: http');
    const xhr = new XMLHttpRequest();
    xhr.addEventListener('load', () => {
      if (xhr.readyState === 4 && xhr.status === 200) {
        observer.next(JSON.parse(xhr.responseText));
        observer.complete();
      }
    });
    xhr.open('GET', url);
    xhr.send();
    
    return () => xhr.abort()
  }
  
  return new Observable(subscriptionFn);
}

Escribimos una función httpque recibe una URL, ejecuta una solicitud http y devuelve una secuencia que emite la respuesta http recibida.

Ahora, mirando nuestra implementación, ¿qué crees que sucederá cuando nos suscribamos a esta transmisión dos veces?


// A small observer helper
const observer = tag => ({
  next(value) {
    console.log(`${tag}:`, value);
  }
});

http('https://jsonplaceholder.typicode.com/users')
  .subscribe(observer('subscriber-1'));

http('https://jsonplaceholder.typicode.com/users')
  .subscribe(observer('subscriber-2'));

Correctamente, se ejecutarán dos solicitudes http. Si observamos nuevamente la implementación de la clase Observable, veremos por qué es así. Cada suscriptor llama a una devolución de llamada de suscripción, que a su vez realiza una solicitud http cada vez.

Operadores

Un operador es una función que toma una secuencia como entrada, realiza cualquier acción y devuelve una secuencia.

Escribiremos nuestro primer operador propio.

function map(fn) {
  return source => {
    return new Observable(observer => {
      log('Observable execution: map');
      return source.subscribe({
        next(value) {
          observer.next(fn(value));
        }
      });
    });
  };
}

La función map()devuelve un operador que acepta la secuencia original y devuelve una secuencia en la que todos los valores de paso se pasarán a través de la función fn.

Aquellos. dentro de él siempre hay una suscripción a la secuencia de entrada.

Antes de usar este nuevo operador, necesitamos adjuntarlo de alguna manera a la transmisión. Extender nuestra clase Observableporpipe()

class Observable {
  constructor(subscriptionFn) {
    this.subscriptionFn = subscriptionFn;
  }

  subscribe(observer) {
    return this.subscriptionFn(observer);
  }

  pipe(...operators) {
    return operators.reduce((source, next) => next(source), this);
  }
}

Un método simple, solo una línea de código. pipeToma una serie de operadores y los llama a su vez, pasando a cada entrada el resultado del anterior.

Usemos nuestro operador:

http('https://jsonplaceholder.typicode.com/users')
  .pipe(map(res => res[0]))
  .subscribe(observer('subscriber'));

Cuando se llama subscribe, se ejecutará una suscripción a la secuencia de salida map()y, a su vez map, se ejecutará una suscripción a la secuencia original en su interior .

La secuencia http emite el valor en el que cae map. Luego, se ejecuta la función fn, la secuencia de mapemite el valor a la suscripción final. Funciona como una observable chaincadena de hilos.

Si nos suscribimos a una cadena dos veces, cada suscripción en la cadena se llamará dos veces.

const firstUser$ = http('https://jsonplaceholder.typicode.com/users')
    .pipe(map(res => res[0]));

firstUser$.subscribe(observer('subscriber-1'));
firstUser$.subscribe(observer('subscriber-2'));

¿Y si este comportamiento no nos conviene? Si queremos llamar a la función de suscripción solo una vez, ¿cuántas suscripciones tendríamos?

Por ejemplo, ¿qué pasa si queremos hacer una solicitud http y usar el resultado para todos los suscriptores? En este caso lo necesitas Subject.

Asignaturas

SubjectEs tanto una transmisión como un suscriptor. El flujo - porque tiene un método subscribe(), el suscriptor - porque implementa la interfaz del suscriptor - métodos next(), error(), complete().

Vamos a escribirlo.

class Subject extends Observable {
  constructor() {
    super();
    this.observers = [];
  }

  subscribe(observer) {
    this.observers.push(observer);
  }

  next(value) {
    this.observers.forEach(observer => observer.next(value));
  }

  error(error) {
    this.observers.forEach(observer => observer.error(error));
  }

  complete() {
    this.observers.forEach(observer => observer.complete());
  }
}

Subjectpuede actuar como intermediario entre la transmisión en frío y muchos suscriptores.

Cambie nuestro ejemplo de la siguiente manera:

const subject = new Subject();
subject.subscribe(observer('subscriber1'));
subject.subscribe(observer('subscriber2'));

http('https://jsonplaceholder.typicode.com/users')
  .pipe(map(res => res[0]))
  .subscribe(subject);

Cuando se llama subject.subscribe(someFn), solo se realiza una operación simple: agregar una subject.observersfunción a la matriz someFn.

Bueno, entonces, dado que también se Subjectcomporta como suscriptor, puede suscribirse a la transmisión original, es decir cuando el hilo original emite un valor, se llama subject.next(), lo que implica la transferencia de este valor a cada uno de los suscriptores subject.

Ahora tenemos que la devolución de llamada original de la suscripción se ejecuta una vez, y solo se ejecutará una solicitud http.

Llegados tarde a la fiesta

¿Qué sucede si la transmisión original ya funcionó antes de registrarnos?

No será posible mostrar esto en el ejemplo anterior, ya que http es asíncrono, incluso si se suscribe inmediatamente después, el valor seguirá apareciendo después de la suscripción.

Creemos rápidamente una función generadora of:


function of(...values) {
  return new Observable(observer => {
    log('Observable execution: of');
    values.forEach(value => observer.next(value));
  });
}

Una secuencia creada por medios of()emite valores sincrónicamente, uno después del otro. Nos suscribiremos subjectdespués de que ya se haya suscrito a.

const subject = new Subject();
of(1, 2, 3).subscribe(subject);

subject.subscribe(observer('subscriber1'));
subject.subscribe(observer('subscriber2'));

Nuestros suscriptores no han recibido nada. ¿Por qué? Nuestra implementación no admite suscriptores "tardíos". Cuando la secuencia original de of()emite valores, los suscriptores aún no están registrados, estos valores no irán a ninguna parte.

En ejemplos reales en Angular, puede ser que la secuencia de origen funcionó, pero su componente aún no está presente en la página. Y cuando aparece el componente, se suscribe a la fuente, pero no recibe los valores que ya han pasado.

Una forma de resolver el problema es esta ReplaySubject. Esbozamos su versión y vemos cómo funciona.


class ReplaySubject extends Subject {
  constructor(bufferSize) {
    super();
    this.observers = [];
    this.bufferSize = bufferSize;
    this.buffer = [];
  }

  subscribe(observer) {
    this.buffer.forEach(val => observer.next(val));
    this.observers.push(observer);
  }

  next(value) {
    if (this.buffer.length === this.bufferSize) {
      this.buffer.shift();
    }

    this.buffer.push(value);
    this.observers.forEach(observer => observer.next(value));
  }
}

El concepto es simple. Como su nombre lo indica, ReplaySubjecteste es Subjectuno especial que puede reproducir valores antiguos a todos los suscriptores nuevos.

Cada valor liberado se transferirá a todos los suscriptores actuales y se guardará para los futuros, el tamaño del búfer se bufferSizeestablece en el constructor.

Reescribe el ejemplo anterior con ReplaySubject.

const subject = new ReplaySubject(3);
of(1, 2, 3).subscribe(subject);

subject.subscribe(observer('subscriber1'));
subject.subscribe(observer('subscriber2'));

El resultado ha cambiado.

A pesar de la suscripción tardía, los atrapamos a todos.

En resumen, el propósito ReplaySubjectes la distribución de valores a todos los suscriptores y almacenarlos en caché para futuros suscriptores "tardíos".

Antes de continuar, le recomiendo que intente escribir su propia implementación BehaviorSubject. Puede encontrar el código terminado al final del artículo.

Ahora finalmente pasamos a operadores de multidifusión. Espero que los ejemplos anteriores te ayuden a entenderlos más rápido.

Operadores de multidifusión

Multicast y Connect

El operador multicast() utiliza Subjectpara emitir la secuencia de origen a varios suscriptores.


import { interval, Subject, ConnectableObservable } from 'rxjs';
import { multicast } from 'rxjs/operators';

const connectableObservable = interval(1000).pipe(
  multicast(new Subject())
)

const observer1 = connectableObservable.subscribe(log);
const observer2 = connectableObservable.subscribe(log);

const connectableSubscription = (connectableObservable as ConnectableObservable<any>)
  .connect();

multicastdevuelve un objeto ConnectableObservableque tiene un método connect. Su propósito es suscribir el asunto recibido a la secuencia fuente.

El método connectnos permite determinar cuándo comenzar la ejecución del hilo original. Hay un momento a tener en cuenta: para darse de baja de la fuente que necesita hacer:

connectableSubscription.unsubscribe();

No estamos limitados a lo simple Subject. En su lugar, puede usar cualquier clase derivada, por ejemplo ReplaySubject:

import { interval, ReplaySubject, ConnectableObservable } from 'rxjs';
import { multicast } from 'rxjs/operators';

const connectableObservable = interval(1000).pipe(
  multicast(new ReplaySubject(1))
)

const observer1 = connectableObservable.subscribe(log);

setTimeout(() => {
  // Late subscriber
  connectableObservable.subscribe(log);
}, 3000)

const connectable = (connectableObservable as ConnectableObservable<any>).connect();

A partir de este código, puede adivinar lo que sucederá debajo del capó.

Cuando usamos multicast, podemos transferir no solo Subject, sino también una función de fábrica, que devuelve una nueva cada vez Subject.

Reutilizado ya completado subjectno puede ser, la función de fábrica resuelve este problema.

interval(1000).pipe(
  multicast(() => new Subject())
)

Refundir

Cuando usamos el operador multicast(), somos responsables de la llamada connect()para iniciar la ejecución del observable original. Además, aún tenemos que monitorear posibles fugas de memoria, cancelando la suscripción manualmente ConnectableSubscription.

La automatización del proceso evitaría errores y simplificaría el código. Los amables desarrolladores de RxJS lo pensaron por nosotros y crearon un refCountoperador.

refCountcuenta las suscripciones y cuando aparece la primera, llama connect(), es decir se suscribe Cuando disminuye de nuevo a cero, se llamará a una respuesta.

const source = interval(1000).pipe(
  multicast(new Subject()),
  refCount()
)
 
// refCount === 1 => source.subscribe();
const observer1 = source.subscribe(log);

// refCount === 2
const observer2 = source.subscribe(log);

setTimeout(() => {
  // refCount - 1
  observer1.unsubscribe();
  // refCount - 1
  observer2.unsubscribe();
  // refCount === 0 => source.unsubcribe();
}, 3000)

Tenga en cuenta que después de refCountobtener el observable habitual, no ConnectableObservable.

Publicar y sus opciones

multicast() + Subject + refCount()Este es un caso bastante típico en RxJS y los desarrolladores lo han reducido a un solo operador.

Veamos qué opciones tenemos.

publish() equivalente multicast(() => new Subject())

const connectableObservable = interval(1000).pipe(
  publish()
)

connectableObservable.connect();

publishBehavior() equivalente multicast(new BehaviorSubject())

const connectableObservable = interval(1000).pipe(
  publishBehavior(100)
)

connectableObservable.connect();

publishReplay() equivalente multicast(() => new ReplaySubject(x))

const connectableObservable = interval(1000).pipe(
  publishReplay(3)
)

connectableObservable.connect();

publishLast() equivalente multicast(new AsyncSubject())

const connectableObservable = interval(1000).pipe(
  take(2),
  publishLast()
)

connectableObservable.connect();

share() equivalente multicast(() => new Subject()) + refCount()
```
const source = interval(1000).pipe(
  share()
)
```
shareReplay(bufferSize) Este es un operador de multidifusión que utiliza ReplaySubject. No tiene adentro multicast()y su resultado es observable, no ConnectableObservable. Se puede usar con refCounto sin él. Aquí hay dos opciones:
```
interval(1000).pipe(
  shareReplay({ refCount: true, bufferSize: 1 })
)

interval(1000).pipe(
  shareReplay(1)
)
```

Cuando se shareReplayllama con { refCount: false }es como llamar shareReplay(x).

En este caso, no habrá recuento de referencias. Esto significa que hasta que se complete la transmisión original, se shareReplaysuscribirá a ella, independientemente de si tiene los suscriptores finales o no. Todos los nuevos suscriptores recibirán los últimos valores de x.

shareReplay vs PublicarReplay + refCount

A primera vista , es shareReplay({ refCount: true, bufferSize: X })idéntico publishReplay(X) + refCount() , pero esto no es del todo cierto.

Veamos cuáles son las similitudes y cuál es la diferencia.

Tienen el mismo comportamiento refCount: suscribirse y darse de baja de la transmisión original en función del número de suscriptores. También reaccionan de la misma manera cuando se completa la transmisión original: todos los suscriptores nuevos reciben X últimos valores.

Sin embargo, si el flujo original aún no está finalizado, en este caso cuando lo hayamos hecho publishReplay(X) + refCount(), todos los suscriptores nuevos recibirán valores X del búfer y luego se volverán a firmar con el mismo ReplaySubject.
Pero si usamos los shareReplay({ refCount: true, bufferSize: 1 })últimos valores de X, no lo obtendrán, ya que en su interior crea uno nuevo ReplaySubject y lo usa para volver a suscribirse a la fuente.

Ejemplos que ilustran esto:

const source = interval(1000).pipe(
  publishReplay(1),
  refCount()
);

const one = source.subscribe(observer('subcriber-1'));

setTimeout(() => {
  one.unsubscribe();
 
  // This subscriber will get the last emitted values from the source
  const two = source.subscribe(observer('subcriber-2'));
}, 3000);

const source = interval(1000).pipe(
  shareReplay({ refCount: true, bufferSize: 1 })
);

const one = source.subscribe(observer('subcriber-1'));

setTimeout(() => {
  one.unsubscribe();
  
  // This subscriber will NOT get the last emitted values from the source
  const two = source.subscribe(observer('subcriber-2'));
}, 3000);

Ejemplos reales en angular

Veamos cómo usar los operadores de multidifusión estudiados en condiciones de combate.

Usamos compartir

Supongamos que tenemos un componente que necesita datos de la secuencia original. Podría ser una solicitud http, un estado o lo que sea. Y también necesitamos manipulación de datos, como filtrado, clasificación, etc.

@Component({
  template: `
    <users-list [users]="allUsers$ | async"></users-list>
  `,
})
export class UsersPageComponent {
  allUsers$: Observable<User[]>;

  constructor(private http: HttpClient) {
  }

  ngOnInit() {
    this.allUsers$ = this.http.get('https://api/users').pipe(
      map(users => filter/sort),
    );
  }
}

Y ahora necesitamos otro componente que muestre solo el primer usuario. Si nos suscribimos a la secuencia de origen tal como está, entonces:

@Component({
  template: `
    <user [user]="firstUser$ | async"></user>
    <users-list [users]="allUsers$ | async"></users-list>
  `,
})
export class UsersPageComponent {
  allUsers$: Observable<User[]>;
  firstUser$: Observable<User>;
  
  constructor(private http: HttpClient) {
  }

  ngOnInit() {
    this.allUsers$ = this.http.get('https://api/users').pipe(
      map(users => filter/sort),
    );
    
    this.firstUser$ = this.allUsers$.pipe(map(users => users[0]));
  }
}

Y ahora tenemos dos solicitudes http, las operaciones de clasificación o filtrado se realizarán dos veces.
Aplicamos share:

@Component({
  template: `
    <user [user]="firstUser$ | async"></user>
    <users-list [users]="allUsers$ | async"></users-list>
  `,
})
export class UsersPageComponent {
  allUsers$: Observable<User[]>;
  firstUser$: Observable<User>;
  
  constructor(private http: HttpClient) {
  }

  ngOnInit() {
    this.allUsers$ = this.http.get('https://api/users').pipe(
      map(users => filter/sort),
      share()
    );
    
    this.firstUser$ = this.allUsers$.pipe(map(users => users[0]));
  }
}

Ya sabemos que crea Subjectuno nuevo que se suscribe a la fuente. Cuando se emite la fuente, el sujeto pasa este valor a todos sus suscriptores.

El problema está resuelto y, cuando nos suscribimos firstUser$, nos suscribimos subjectdirectamente a la transmisión interna y no a la transmisión original.

Usando ShareReplay

ShareReplayse aplica cuando necesita emitir, almacenar en caché y repetir los últimos valores de X. Un ejemplo típico es un servicio singleton que realiza una solicitud http.


@Injectable({ providedIn: 'root' })
export class BlogService {
  posts$ = this.http.get('https://jsonplaceholder.typicode.com/posts')
              .pipe(shareReplay(1));

  constructor(private http: HttpClient) {}
}

No importa cuántos componentes solicitarán datos ahora o en el futuro, solo habrá una solicitud http y el resultado se guardará en el búfer interno ReplaySubject.
Todavía puede haber un caso en el que deba cancelar una solicitud incompleta, ya que no hay suscriptores, entonces deberá presentar una solicitud refCount.

El código completo se puede encontrar aquí .

Estudiamos operadores RxJS multicast