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

user_583810 0
Viewed 1199

If I have an element in my template which is using property binding, bound to an observable, using async pipe to ensure it is unsubscribed automatically...then would using takeUntil(this.destroy$) be unnecessary?

Template snippet:

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

Component:

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();
  }    
}
rxjs-observables rxjs angular typescript
1 Answers

TL;DR - No need for that because the async pipe does that automatically when the component gets destroyed.

The details

As mentioned in 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.

Here is the source code of how the async pipe is handling it:

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

Where the used strategy is one of two classes that implement the SubscriptionStrategy interface:

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 {}
}
user_151596030
Related