feat(common/pipes/async_pipe): implement async pipe for $q and RxJS 5…

… observables

Closes #98

common.ts

@@ -1,3 +1,4 @@
 //import {} from './src/common/services';
 
 export * from './src/common/directives';
+export * from './src/common/pipes';

src/common/pipes.ts

@@ -0,0 +1,8 @@
+/**
+ * @module
+ * @description
+ * This module provides a set of common Pipes.
+ */
+
+export { AsyncPipe } from './pipes/async_pipe';
+export { COMMON_PIPES } from './pipes/common_pipes';

src/common/pipes/async_pipe.ts

@@ -0,0 +1,88 @@
+import { Subscription } from 'rxjs/Subscription';
+import { Observable } from 'rxjs/Observable';
+
+
+import { Pipe } from '../../core/pipes/decorators';
+import { PipeTransform } from '../../core/pipes/pipe_interfaces';
+import { isPromiseLike } from '../../facade/lang';
+import { isObservable } from '../../facade/lang';
+/**
+ * Thanks to @cvuorinen for the angular1-async-filter
+ * https://github.com/cvuorinen/angular1-async-filter
+ */
+
+type StoredSubscription = ng.IPromise<any>|ng.IHttpPromise<any>|Subscription;
+
+@Pipe( { name: 'async'/*, pure: false*/ } )
+export class AsyncPipe implements PipeTransform {
+
+  // Need a way to tell the input objects apart from each other (so we only subscribe to them once)
+  private static nextObjectID: number = 0;
+  private static values: {[key: string]: any} = {};
+  private static subscriptions: {[key: string]: StoredSubscription} = {};
+
+  private static TRACK_PROP_NAME = '__asyncFilterObjectID__';
+
+  private static objectId( obj: any ): any {
+    if ( !obj.hasOwnProperty( AsyncPipe.TRACK_PROP_NAME ) ) {
+      obj[ AsyncPipe.TRACK_PROP_NAME ] = ++AsyncPipe.nextObjectID;
+    }
+    return obj[ AsyncPipe.TRACK_PROP_NAME ];
+  }
+
+  private static isPromiseOrObservable( obj: any ): boolean {
+    return isPromiseLike( obj ) || isObservable( obj );
+  }
+
+  private static getSubscriptionStrategy( input: any ): ( value ) => StoredSubscription {
+    return input.subscribe && input.subscribe.bind( input )
+      || input.success && input.success.bind( input ) // To make it work with HttpPromise
+      || input.then.bind( input ); // To make it work with Promise
+  }
+
+  private static dispose( inputId: number ): void {
+    if ( AsyncPipe.subscriptions[ inputId ] && (AsyncPipe.subscriptions[ inputId ] as Subscription).unsubscribe ) {
+      (AsyncPipe.subscriptions[ inputId ] as Subscription).unsubscribe();
+    }
+    delete AsyncPipe.subscriptions[ inputId ];
+    delete AsyncPipe.values[ inputId ];
+  }
+
+  transform( input: Observable<any>|ng.IPromise<any>|ng.IHttpPromise<any>, scope?: ng.IScope ): any {
+
+    if ( !AsyncPipe.isPromiseOrObservable( input ) ) {
+      return input
+    }
+
+    const inputId = AsyncPipe.objectId( input );
+
+    // return cached immediately
+    if ( inputId in AsyncPipe.subscriptions ) {
+      return AsyncPipe.values[ inputId ] || undefined;
+    }
+
+    const subscriptionStrategy = AsyncPipe.getSubscriptionStrategy( input );
+    AsyncPipe.subscriptions[ inputId ] = subscriptionStrategy( _setSubscriptionValue );
+
+    if ( scope && scope.$on ) {
+      // Clean up subscription and its last value when the scope is destroyed.
+      scope.$on( '$destroy', () => { AsyncPipe.dispose( inputId ) } );
+    }
+
+    function _setSubscriptionValue( value: any ): void {
+      AsyncPipe.values[ inputId ] = value;
+
+      // this is needed only for Observables
+      _markForCheck( scope );
+    }
+
+    function _markForCheck( scope: ng.IScope ) {
+      if ( scope && scope.$applyAsync ) {
+        // @TODO perfmatters we don't need to digest whole scope tree right?
+        // scope.$applyAsync(); // Automatic safe apply, if scope provided
+        scope.$digest(); // Automatic safe apply, if scope provided
+      }
+    }
+
+  }
+}

src/common/pipes/common_pipes.ts

@@ -0,0 +1,19 @@
+/**
+ * @module
+ * @description
+ * This module provides a set of common Pipes.
+ */
+import { AsyncPipe } from './async_pipe';
+
+/**
+ * A collection of Angular core pipes that are likely to be used in each and every
+ * application.
+ *
+ * This collection can be used to quickly enumerate all the built-in pipes in the `pipes`
+ * property of the `@Component` decorator.
+ *
+ * @experimental Contains i18n pipes which are experimental
+ */
+export const COMMON_PIPES = [
+  AsyncPipe
+];

src/facade/lang.ts

@@ -133,6 +133,14 @@ export function isPromise(obj: any): boolean {
   return obj instanceof (<any>_global).Promise;
 }
 
+export function isPromiseLike( obj: any ): boolean {
+  return Boolean( isPresent( obj ) && obj.then );
+}
+
+export function isObservable( obj: any ): boolean {
+  return Boolean( isPresent( obj ) && obj.subscribe );
+}
+
 export function isJsObject( o: any ): boolean {
   return o !== null && (typeof o === "function" || typeof o === "object");
 }