RxJS: takeUntil() Angular компонент ngOnDestroy()

Ответ 1

Вы можете использовать ReplaySubject для этого:

РЕДАКТИРОВАТЬ: отличается от RxJS 6.x: обратите внимание на использование метода 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();
  }
}

Это действительно только для RxJS 5.x и старше:

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

Ответ 2

Использование функции componentDestroyed() из пакета npm @w11k/ngx-componentdestroyed - безусловно, самый простой способ использования 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() {}
}

Вот версия componentDestroyed() для непосредственного включения в ваш код:

// 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$;
}

Ответ 3

Ну, это сводится к тому, что вы подразумеваете, закрывая подписку. Есть два способа сделать это:

• Использование оператора, который завершает цепочку (например, takeWhile()).
• Отписаться от источника Observable.

Хорошо знать, что эти два не совпадают.

При использовании, например, takeWhile() вы отправляете оператору сообщение complete, которое распространяется на ваших наблюдателей. Поэтому, если вы определяете:

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

Затем, когда вы завершите цепочку, например. takeWhile() будет вызываться функция doWhatever().

Например, он может выглядеть так:

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

Через 3 секунды будут вызываться все обратные вызовы.

С другой стороны, когда вы отказываетесь от подписки, вы говорите, что вас больше не интересуют элементы, созданные источником Observable. Однако это не означает, что источник должен завершиться. Вам просто все равно.

Это означает, что вы можете собрать все вызовы Subscription из .subscribe(...) и сразу отказаться от них:

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

Теперь после 3-секундной задержки ничего не будет напечатано на консоль, потому что мы отменили подписку и не был вызван полный обратный вызов.

Итак, что вы хотите использовать, зависит от вас и вашего прецедента. Просто имейте в виду, что отмена подписки - это не то же самое, что завершить, даже если я полагаю, что в вашей ситуации это не имеет особого значения.

Ответ 4

Если ваш компонент напрямую связан с маршрутом, вы можете избежать добавления состояния, используя события Router с помощью takeUntil(). Таким образом, как только вы отойдете от компонента, он автоматически очистит свои подписки.

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 => {
                ...
            });
    }
}

Примечание. Этот простой пример не учитывает охраняемые маршруты или отмененную навигацию по маршруту. Если существует вероятность того, что одно из событий маршрутизатора может быть запущено, но навигация по маршруту отменена, вам нужно будет отфильтровать события маршрутизатора, чтобы они инициировались в соответствующей точке - например, после проверки Route Guard или после навигации завершено.

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