Errores comunes de RxJS en Angular

Pretende señalar errores comunes y sencillos, explicándolos y resolviéndolos brevemente.

No establezcas valores de sincronización con las suscripciones en Angular

❌ No lo hagas:

service.getFoo$().subscribe(v => {
  this.foo = foo
});

✅ Hacer:

this.foo$ = service.getFoo$();

❓ Por qué:

Esto mezcla código async y sync, introduciendo race conditions.

Muchas veces código como este pasa a funcionar cuando se escribe porque los valores observables pueden ser emitidos de forma sincrónica o asincrónica, incluso por el mismo observable.

Por ejemplo, el almacén NgRX emite el valor actual de forma síncrona, pero los valores posteriores son asíncronos.

Además, la detección de cambios de Angular no controlará tus suscripciones manuales.
Así que si estás usando ese valor en el HTML, puede que no veas el nuevo valor renderizado.
Este es un problema tan común que Angular lanza el error "NG0100: Expression has changed after it was checked".

El AsyncPipe activa la detección de cambios:

<div>{{ foo$ | async}}</div> 

"Gestiona" siempre tus suscripciones

❌ No lo hagas:

someObservable$.subscribe(...);

✅ Hacer:

 subscriptions: Subscription[];

ngOnDestroy() {
  this.subscriptions.forEach(s => s.unsubsribe())
}

...
this.subscriptions.push(someObsevable$.subscribe(...));
// and/or cleanup another way, suchs as take/takeUntil/takeWhile etc.


// Or, angular 16+
someObservable.pipe(takeUntilDestroyed()).subscribe(...);

❓ Por qué:

Fuga de memoria

Una suscripción no será recolectada cuando se espera que lo sea (es decir, cuando se destruye el componente).

Esto significa que las suscripciones pueden permanecer para siempre (y evitar que lo que sea que esté apuntando sea recolectado).

Incluso si estás seguro de que no habrá más valores, el suscriptor no puede saberlo - a menos que el observador se complete explícitamente .
Y si hay más valores, la suscripción se ejecutará de nuevo incluso si no es tu intención.

Edit: El takeUntilDestroyed de Angular 16 simplifica este proceso.

Recuerda ponerlo como último (o penúltimo) operador.

Notas:
- Frecuentemente, deberías incluso limpiar antes de que el componente sea destruido. Ve a la sección sobre limitar sus observables.
- El array de suscripciones no es el único método. Ni siquiera es mi favorito. Es sólo la forma más común en nuestra base de código actualmente. Incluso tenemos una clase BaseComponent que simplemente heredamos para ello.
- Edit: estamos adoptando el nuevo takeUntilDestroyed mencionado anteriormente.
- AsyncPipe de Angular se encargará de la limpieza por sí mismo. Otra razón para preferirlo a las suscripciones manuales.
A veces, quieres seguir ejecutando suscripciones incluso mientras se destruye el componente. En ese caso, no lo limpies al destruirlo. Casi siempre querrás limitar el observable en estos casos.
Por ejemplo, ejecutar una llamada HTTP al cerrar un diálogo:

submit() {
 this.someService.doHttpRequest(this.form.value).pipe(take(1)).subscribe();
 this.dialogRef.close();
}

Considera la posibilidad de limitar sus observables

✅ Hacer:

Utiliza first, take, takeUntil, takeWhile y otros "limitadores".

❓ Por qué:

Con frecuencia, hay un número limitado de veces que queremos ejecutar una suscripción.
A veces, sería un error no hacerlo:

// Infinite loop
store.select(getCurrentFoo)
  // Here there should be a `.pipe(take(1))`
  .subscribe((foo) => {
    const newFoo = createNewFoo(foo);
    // This will update `foo` in the store, which is what we're observing,
    // so we end up with an infinite loop
    store.dispatch(UpdateFoo(newFoo));
  });

A veces, es simplemente una optimización similar a la anterior, pero nunca habrá otro valor.
Podemos limpiar la suscripción en algún momento (por ejemplo, al destruir un componente), pero eso podría llevar mucho tiempo:

ngOnInit() { 
  this.subsriptions.push(
    store.select(getProjectId)
      // Here there should be a `.pipe(take(1))`
      .subscribe(projectId => {
        ...
      })
  );
}

En el ejemplo anterior, sabemos que si el projectId cambia habrá una redirección, y el componente se destruirá de todos modos, por lo tanto podemos gestionar esto correctamente.

Sin embargo, RxJS no lo sabe. La suscripción se sentará y esperará un nuevo projectId que nunca llega, mientras el componente esté abierto.

ℹ️ Algunos observables sólo tienen 1 emisión, por ejemplo las peticiones HTTP. take(1) es técnicamente innecesario en estos casos, sin embargo puede ser más robusto en caso de que se produzcan cambios, y "no hace daño".

Considera la multidifusión 🪄 pero ten cuidado 😨.

tl;dr
La multidifusión es un poco confusa, así que no lo compartas todo a ciegas.
En su lugar, úsalo cuando tengas una razón concreta - típicamente peticiones HTTP repetidas como en el ejemplo de abajo.
⚠️ Cuando se utiliza shareReplay, por lo general debe ser configurado con refCount: true.

❓ Qué:

Múltiples suscripciones = múltiples ejecuciones; (Y para tu información: cada AsyncPipe es una suscripción).

Si la lógica observable es cara (lenta o consume muchos recursos), ese gasto se multiplica.

Ejemplo:

this.foo$ = fooHttpService.get(fooId);
this.subscriptions.push(
  this.foo$.subscribe(...), // 1st subscription
  bar$.pipe(withLatestFrom(foo$).subscribe(...), // 2nd subscription
);

...

<div *ngIf="foo$ | async as foo"> // 3rd subscription

Abre tu consola. Hay 3 peticiones http.

Solución: utiliza shareReplay():

this.foo$ = fooHttpService.get(fooId)
  .pipe(shareReplay({ bufferSize: 1, refCount: true }));

⚠️ shareReplay requiere una gestión especial

Observa que está configurado específicamente con refCount: true. Esto se debe a que, de lo contrario, tendrá una suscripción no gestionada.

Cuando te suscribes a shareReplay, se crea una única suscripción interna para compartir entre todos los clientes.

Con refCount, shareReplay cancelará internamente la suscripción cuando todos los clientes se hayan dado de baja.

Es conceptualmente similar a una referencia débil, si estás familiarizado con ellas.
El inconveniente es que si la siguiente suscripción se produce después de que el anterior se haya dado de baja, el shareReplay ya se habrá "reiniciado" entre ellos, por lo que no se habrá compartido nada.

Sin refCount, la suscripción interna permanecerá, para evitar el trabajo computacionalmente costoso para cualquier suscripción futura  como una caché que no se limpia.
Cancelar la suscripción del observable resultante no cancelará la suscripción interna.
Si el observable "fuente" se completa de todos modos, no pasa nada, sin embargo es delicado.
Aquí es donde nuestro documento interno anima a buscar el consejo de otros miembros del equipo.

ℹ️ shareReplay no es el único operador de multidifusión, pero es el que probablemente más te interese.

ℹ️ En RxJS 7, los operadores de multidifusión fueron refactorizados masivamente. En el momento de escribir este artículo, estamos en la v6.

ℹ️ Vale la pena mencionarlo: compartir tiene una relación especial con limitar.

Fuente