RxJS: takeUntil() ngOnDestroy() del componente angular

unsubscribe observable angular 4

Question

tl; dr: Básicamente quiero casarme con ngOnDestroy de ngOnDestroy con el operador takeUntil() . -- ¿es eso posible?

Tengo un componente angular que abre varias suscripciones Rxjs. Estos deben cerrarse cuando se destruye el componente.

Una solución simple para esto sería:

class myComponent {
private subscriptionA;
private subscriptionB;
private subscriptionC;
constructor(
private serviceA: ServiceA,
private serviceB: ServiceB,
private serviceC: ServiceC) {}
ngOnInit() {
this.subscriptionA = this.serviceA.subscribe(...);
this.subscriptionB = this.serviceB.subscribe(...);
this.subscriptionC = this.serviceC.subscribe(...);
}
ngOnDestroy() {
this.subscriptionA.unsubscribe();
this.subscriptionB.unsubscribe();
this.subscriptionC.unsubscribe();
}
}

Esto funciona, pero es un poco redundante. Especialmente no me gusta eso: la unsubscribe() la unsubscribe() está en otra parte, así que debes recordar que están vinculados. - El estado del componente está contaminado con la suscripción.

Preferiría mucho usar el operador takeUntil() o algo similar, para que se vea así:

class myComponent {
constructor(
private serviceA: ServiceA,
private serviceB: ServiceB,
private serviceC: ServiceC) {}
ngOnInit() {
const destroy = Observable.fromEvent(???).first();
this.subscriptionA = this.serviceA.subscribe(...).takeUntil(destroy);
this.subscriptionB = this.serviceB.subscribe(...).takeUntil(destroy);
this.subscriptionC = this.serviceC.subscribe(...).takeUntil(destroy);
}
}

¿Hay un evento de destrucción o algo similar que me permita usar takeUntil() u otra forma de simplificar la arquitectura de componentes de esa manera? Me doy cuenta de que podría crear un evento yo mismo en el constructor o algo que se desencadene en ngOnDestroy() pero que al final no haría las cosas mucho más ngOnDestroy() de leer.

Answer 1

Bueno, esto se reduce a lo que quiere decir al cerrar una suscripción. Hay básicamente dos formas de hacer esto:

1. Usar un operador que complete la cadena (como takeWhile() ).
2. Darse de baja de la fuente observable.

Es bueno saber que estos dos no son lo mismo.

Cuando utiliza, por ejemplo, takeWhile() , hace que el operador envíe complete notificación complete que se propaga a sus observadores. Así que si define:

...
.subscribe(..., ..., () => doWhatever());

Luego cuando termine la cadena con eg. takeWhile() doWhatever() función doWhatever() .

Por ejemplo, podría verse así:

const Observable = Rx.Observable;
const Subject = Rx.Subject;
let source = Observable.timer(0, 1000);
let subject = new Subject();
source.takeUntil(subject).subscribe(null, null, () => console.log(''complete 1''));
source.takeUntil(subject).subscribe(null, null, () => console.log(''complete 2''));
source.takeUntil(subject).subscribe(null, null, () => console.log(''complete 3''));
setTimeout(() => {
subject.next();
}, 3000);

Después de 3s se llamarán todos los callbacks completos.

Por otro lado, cuando cancela la suscripción, está diciendo que ya no está interesado en los elementos producidos por la fuente Observable. Sin embargo, esto no significa que la fuente tiene que completar. Simplemente no te importa más.

Esto significa que puede recopilar todas las Subscription de las .subscribe(...) y cancelar la suscripción a todas de una vez:

let subscriptions = new Rx.Subscription();
let source = Observable.timer(0, 1000);
subscriptions.add(source.subscribe(null, null, () => console.log(''complete 1'')));
subscriptions.add(source.subscribe(null, null, () => console.log(''complete 2'')));
subscriptions.add(source.subscribe(null, null, () => console.log(''complete 3'')));
setTimeout(() => {
subscriptions.unsubscribe();
}, 3000);

Ahora, después de 3s delay, no se imprimirá nada en la consola porque cancelamos la suscripción y no se invocó ninguna devolución de llamada completa.

Entonces, lo que quieras usar depende de ti y de tu caso de uso. Solo ten en cuenta que cancelar la suscripción no es lo mismo que completar aunque supongo que en tu situación no importa.

Answer 2

El uso de la función componentDestroyed() del paquete ng2-rx-componentdestroyed es, con mucho, la forma más fácil de usar takeUntil:

@Component({
selector: ''foo'',
templateUrl: ''./foo.component.html''
})
export class FooComponent implements OnInit, OnDestroy {
ngOnInit() {
Observable.interval(1000)
.takeUntil(componentDestroyed(this)) // <--- magic is here!
.subscribe(console.log);
}
ngOnDestroy() {}
}

Aquí hay una versión de componentDestroyed() para incluir directamente en su código:

// Based on https://www.npmjs.com/package/ng2-rx-componentdestroyed
import { OnDestroy } from ''@angular/core'';
import { ReplaySubject } from ''rxjs/ReplaySubject'';
export function componentDestroyed(component: OnDestroy) {
const oldNgOnDestroy = component.ngOnDestroy;
const destroyed$ = new ReplaySubject<void>(1);
component.ngOnDestroy = () => {
oldNgOnDestroy.apply(component);
destroyed$.next(undefined);
destroyed$.complete();
};
return destroyed$;
}

Answer 3

Podrías aprovechar un ReplaySubject para eso:

EDITAR: Diferente desde RxJS 6.x: tenga en cuenta el uso del método pipe() .

class myComponent {
private destroyed$: ReplaySubject<boolean> = new ReplaySubject(1);
constructor(
private serviceA: ServiceA,
private serviceB: ServiceB,
private serviceC: ServiceC) {}
ngOnInit() {
this.serviceA
.pipe(takeUntil(this.destroyed$))
.subscribe(...);
this.serviceB
.pipe(takeUntil(this.destroyed$))
.subscribe(...);
this.serviceC
.pipe(takeUntil(this.destroyed$))
.subscribe(...);
}
ngOnDestroy() {
this.destroyed$.next(true);
this.destroyed$.complete();
}
}

Esto solo es válido para RxJS 5.x y anteriores:

class myComponentOld {
private destroyed$: ReplaySubject<boolean> = new ReplaySubject(1);
constructor(private serviceA: ServiceA) {}
ngOnInit() {
this.serviceA
.takeUntil(this.destroyed$)
.subscribe(...);
}
ngOnDestroy() {
this.destroyed$.next(true);
this.destroyed$.complete();
}
}

Answer 4

Si su componente está vinculado directamente a una ruta, puede evitar agregar estado al aprovechar los eventos del Router con takeUntil() . De esa manera, tan pronto como se aleje del componente, limpiará automáticamente sus suscripciones.

import { Component, OnInit } from ''@angular/core'';
import { ActivatedRoute, Router } from ''@angular/router'';
import { MyService } from ''./my.service'';
import { Observable } from ''rxjs/Observable'';
import ''rxjs/add/operator/takeUntil'';
@Component({
...
})
export class ExampleComopnent implements OnInit {
constructor(private router: Router, private myService: MyService) {
}
ngOnInit() {
this.myService.methodA()
.takeUntil(this.router.events)
.subscribe(dataA => {
...
});
this.myService.methodB()
.takeUntil(this.router.events)
.subscribe(dataB => {
...
});
}
}

Nota: este ejemplo simple no tiene en cuenta las rutas protegidas o la navegación cancelada de la ruta. Si existe la posibilidad de que se desencadene uno de los eventos del enrutador, pero la navegación de la ruta se cancela, deberá filtrar los eventos del enrutador para que se active en el punto adecuado, por ejemplo, después de la verificación de la Guardia de la Ruta o una vez que navegue Esta completo.

this.myService.methodA()
.takeUntil(this.router.events.filter(e => e instanceOf NavigationEnd))
.subscribe(dataA => {
...
});