Angular RxJS：如果我使用异步管道，我需要执行 .takeUntil(this.destroy$) 吗？

Question

如果我的模板中有一个元素使用属性绑定，绑定到一个可观察对象，使用 async 管道确保它自动取消订阅...那么将使用 takeUntil(this.destroy$) 没必要吗？

模板片段：

<input name="SomeElement"
    [ngModel]="boundObservable$ | async">
</input>

组件：

export class MyComponent implements OnDestroy {
  destroy$: Subject<any> = new Subject<any>();

  boundObservable$ = this.sourceObservable$.pipe(
    takeUntil(this.destroy$),
    map(this.transformData)
  );

  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }    
}

Answer 1

TL;DR - 不需要，因为 async 管道会在组件被销毁时自动执行此操作。

详情

如Angular docs所述：

The async pipe subscribes to an Observable or Promise and returns the latest value it has emitted. When a new value is emitted, the async pipe marks the component to be checked for changes. When the component gets destroyed, the async pipe unsubscribes automatically to avoid potential memory leaks.

这里是 source code async 管道如何处理它：

  // Create Subscription: _subscribe is called from the pipe's `transform` function:
  private _subscribe(obj: Subscribable<any>|Promise<any>|EventEmitter<any>): void {
    this._obj = obj;
    this._strategy = this._selectStrategy(obj);
    this._subscription = this._strategy.createSubscription(
        obj, (value: Object) => this._updateLatestValue(obj, value));
  }

  // Dispose Subscription:
  ngOnDestroy(): void {
    if (this._subscription) {
      this._dispose();
    }
  }

  private _dispose(): void {
    this._strategy.dispose(this._subscription!);
    this._latestValue = null;
    this._subscription = null;
    this._obj = null;
  }

其中使用的策略是实现 SubscriptionStrategy 接口的两个类之一：

interface SubscriptionStrategy {
  createSubscription(async: Subscribable<any>|Promise<any>, updateLatestValue: any): Unsubscribable
      |Promise<any>;
  dispose(subscription: Unsubscribable|Promise<any>): void;
  onDestroy(subscription: Unsubscribable|Promise<any>): void;
}

class SubscribableStrategy implements SubscriptionStrategy {
  createSubscription(async: Subscribable<any>, updateLatestValue: any): Unsubscribable {
    return async.subscribe({
      next: updateLatestValue,
      error: (e: any) => {
        throw e;
      }
    });
  }

  dispose(subscription: Unsubscribable): void {
    subscription.unsubscribe();
  }

  onDestroy(subscription: Unsubscribable): void {
    subscription.unsubscribe();
  }
}

class PromiseStrategy implements SubscriptionStrategy {
  createSubscription(async: Promise<any>, updateLatestValue: (v: any) => any): Promise<any> {
    return async.then(updateLatestValue, e => {
      throw e;
    });
  }

  dispose(subscription: Promise<any>): void {}

  onDestroy(subscription: Promise<any>): void {}
}

Angular RxJS：如果我使用异步管道，我需要执行 .takeUntil(this.destroy$) 吗？

Angular RxJS: Do I need to do .takeUntil(this.destroy$) if I'm using async pipe?

rxjs

typescript

angular

rxjs-observables