fix(async pipe): prevent firing $digest after scope is destroyed.

playground/app/components/async-example/async-destroy.component.ts

@@ -0,0 +1,107 @@
+import 'rxjs/add/operator/take';
+import 'rxjs/add/operator/takeUntil';
+import 'rxjs/add/operator/map';
+import 'rxjs/add/operator/distinctUntilChanged';
+import { Component, Inject, Input, OnDestroy } from 'ng-metadata/core';
+import { BehaviorSubject } from 'rxjs/BehaviorSubject';
+import { Subject } from 'rxjs/Subject';
+
+interface State {
+  data: number;
+}
+
+@Component({
+  selector: 'async-destroy-child',
+  template: (
+    `State is {{ ($ctrl.state$ | async:this).data }}`
+  )
+})
+export class AsyncDestroyChildComponent implements OnDestroy {
+  @Input('<') state$: BehaviorSubject<State>;
+
+  private destroy = false;
+
+  constructor(@Inject('$scope') private $scope: ng.IScope){
+    this.$scope.$watch(() => {
+      // watch digest loops
+      if (this.destroy) {
+        alert('Memory leak');
+      }
+      return true;
+    },() => {});
+  }
+
+  ngOnDestroy() {
+    this.destroy = true;
+  }
+}
+
+@Component({
+  selector: 'async-destroy-container',
+  template: (
+    `<async-destroy-child 
+        state$="$ctrl.state$"></async-destroy-child>`
+  )
+})
+export class AsyncDestroyContainerComponent implements OnDestroy {
+  @Input('<') state$: BehaviorSubject<State>;
+
+  constructor(){}
+
+  ngOnDestroy() {
+    // to start synchronous effects while destroy phase
+    this.state$.next({data: 1});
+  }
+}
+
+@Component({
+  selector: 'async-destroy',
+  template: (
+    `<hl>
+     <h4>Check for Memory leak</h4>
+     <button ng-click="$ctrl.toggle()">Start</button>
+     <div ng-if="$ctrl.showChild$ | async:this">
+        <async-destroy-container
+          state$="$ctrl.state$"></async-destroy-container>
+    </div>`
+  )
+})
+export class AsyncDestroyComponent implements OnDestroy {
+  state$ = new BehaviorSubject<State>({
+    data: 1
+  });
+
+  // observable to create/destroy underlying component
+  showChild$ = this.state$.map((state: State) => state.data % 2)
+    .distinctUntilChanged();
+
+  private ngOnDestroy$ = new Subject<void>();
+
+  constructor(@Inject('$interval') private $interval: ng.IIntervalService){
+    this.runEffects();
+  }
+
+  ngOnDestroy() {
+    this.ngOnDestroy$.next();
+    this.ngOnDestroy$.complete();
+  }
+
+  toggle() {
+    this.state$.take(1)
+      .subscribe((state: State) => {
+        this.state$.next({...state,
+          data: state.data + 1});
+      });
+  }
+
+  private runEffects() {
+    // simulate some synchronous effects
+    this.state$
+      .takeUntil(this.ngOnDestroy$)
+      .subscribe((state: State) => {
+        if (state.data % 3) { // to prevent infinite loop
+          this.state$.next({...state, data: state.data + 1});
+        }
+      });
+  }
+}

playground/app/components/async-example/async-example.component.ts

@@ -48,6 +48,7 @@ export class AsyncTaskComponent {
       </ul>
       <pre style="overflow:auto;max-height:250px;">{{ $ctrl.repos | async | json }}</pre>
     </div>
+    <async-destroy class="clearfix"></async-destroy>
     `
   )
 })

playground/app/components/async-example/async-example.module.ts

@@ -1,10 +1,11 @@
 import { NgModule } from 'ng-metadata/core';
 import { AsyncExampleComponent, AsyncTaskComponent } from './async-example.component';
+import { AsyncDestroyChildComponent, AsyncDestroyComponent, AsyncDestroyContainerComponent } from './async-destroy.component';
 
 @NgModule({
   imports: [],
   // exports, bootstrap, entryComponents have no real functionality yet
   exports: [AsyncExampleComponent],
-  declarations: [AsyncExampleComponent, AsyncTaskComponent]
+  declarations: [AsyncExampleComponent, AsyncTaskComponent, AsyncDestroyComponent, AsyncDestroyChildComponent, AsyncDestroyContainerComponent]
 })
 export class AsyncExampleModule{}

src/common/pipes/async_pipe.ts

@@ -45,6 +45,7 @@ export class AsyncPipe implements PipeTransform {
   private static nextObjectID: number = 0;
   private static values: {[key: string]: any} = {};
   private static subscriptions: {[key: string]: StoredSubscription} = {};
+  private static timeouts: {[key: string]: number} = {};
   private static TRACK_PROP_NAME = '__asyncFilterObjectID__';
 
   private static _objectId( obj: any ): any {
@@ -60,17 +61,24 @@ export class AsyncPipe implements PipeTransform {
       || input.then.bind( input ); // To make it work with Promise
   }
 
-  private static _markForCheck( scope: ng.IScope ) {
+  private static _markForCheck( scope: ng.IScope, inputId: number ) {
     if ( isScope( scope ) ) {
+      // clear previous timeout before starting new
+      clearTimeout(AsyncPipe.timeouts[inputId]);
       // #perfmatters
       // wait till event loop is free and run just local digest so we don't get in conflict with other local $digest
-      setTimeout( ()=>scope.$digest() );
+      AsyncPipe.timeouts[inputId] = setTimeout( ()=> {
+        scope.$digest();
+      });
       // we can't run local scope.$digest, because if we have multiple async pipes on the same scope 'infdig' error would occur :(
       // scope.$applyAsync(); // Automatic safe apply, if scope provided
     }
   }
 
   private static _dispose( inputId: number ): void {
+    // should prevent digest of destroyed scope
+    clearTimeout(AsyncPipe.timeouts[inputId]);
+    delete AsyncPipe.timeouts[inputId];
     if ( isSubscription( AsyncPipe.subscriptions[ inputId ] ) ) {
       (AsyncPipe.subscriptions[ inputId ] as Subscription).unsubscribe();
     }
@@ -106,7 +114,7 @@ export class AsyncPipe implements PipeTransform {
     function _setSubscriptionValue( value: any ): void {
       AsyncPipe.values[ inputId ] = value;
       // this is needed only for Observables
-      AsyncPipe._markForCheck( scope );
+      AsyncPipe._markForCheck( scope, inputId );
     }
 
   }