Предотвращение утечек с помощью комбайна RXJSПоследний с забором труб

Question

При использовании Rxjs возникает некоторая путаница относительно того, как оператор takeUntil ведет себя с combineLatest при использовании pipe.

Необходимо ли добавить takeUntil к каждому внутреннему наблюдаемому, как это, включаяpiped takeUntil:

import {combineLatest} from 'rxjs';
import {takeUntil} from 'rxjs/operators';

---

combineLatest(
      this._store.pipe(select(stateOfLocation), takeUntil(this._destroy)),
      this._store.pipe(select(stateOfPlacesSelected), takeUntil(this._destroy))
    )
      .pipe(takeUntil(this._destroy$))
      .subscribe( ([location, places])  => {
        /* Business Logic */
      }); 

... или pipe(takeUntil(this._destroy$)) будет обрабатывать внутренние наблюдаемые из combineLatest, например:

combineLatest(
      this._store.pipe(select(stateOfLocation)),
      this._store.pipe(select(stateOfPlacesSelected))
    )
      .pipe(takeUntil(this._destroy$))
      .subscribe( ([location, places])  => {
        /* Business Logic */
      });

Я предполагаю, что позже это правильно, но хочу перепроверить, чтобы убедиться, что я правильно понимаю.

Answer 1

Как правило: вы делаете takeUntil ровно один раз.В этом смысл паттерна: хранить его в одном месте.

После того, как вы onNext(...) на this._destroy$, takeUntil гарантированно сообщит о завершении, что, в свою очередь, вызовет логику разрыва (среди которых отписаться).