RxJS: takeUntil() Angular component's ngOnDestroy()

Question

tl;dr: Basically I want to marry Angular's ngOnDestroy with the Rxjs takeUntil() operator. -- is that possible?

I have an Angular component that opens several Rxjs subscriptions. These need to be closed when the component is destroyed.

A simple solution for this would be:

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();
  }

}

This works, but it's a bit redundant. I especially don't like that - The unsubscribe() is somewhere else, so you gotta remember that these are linked. - The component state is polluted with the subscription.

I would much prefer using the takeUntil() operator or something similar, to make it look like this:

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);
  }

}

Is there a destroy event or something similar that would let me use takeUntil() or another way to simplify the component architecture like that? I realize I could create an event myself in the constructor or something that gets triggered within ngOnDestroy() but that would in the end not make things that much simpler to read.

Answer 1

You could leverage a ReplaySubject for that:

EDIT: Different since RxJS 6.x: Note the use of the pipe() method.

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();
  }
}

This is only valid for RxJS 5.x and older:

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();
  }
}