Merge pull request #103 from ngParty/async-pipe

Async pipe

Closes #98

common.ts

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

playground/app/app.component.html

@@ -164,6 +164,18 @@ <h3>Docs examples:</h3>
   </div>
 </section>
 
+<!--ASYNC PIPE EXAMPLE-->
+<section class="demo-content">
+  <button
+    class="mdl-button mdl-js-button mdl-button--accent"
+    ng-click="$ctrl.showAsyncPipeExample=!$ctrl.showAsyncPipeExample">
+    {{ $ctrl.showAsyncPipeExample ? 'hide' : 'show'}} <code>AsyncPipe</code> example
+  </button>
+  <div ng-if="$ctrl.showAsyncPipeExample">
+    <async-example></async-example>
+  </div>
+</section>
+
 <!--Title-->
 <section class="demo-content">
   <button

playground/app/app.component.ts

@@ -1,10 +1,11 @@
 import { Component, Inject, OnInit } from 'ng-metadata/core';
 import { DynamicValueToken, NgRxStore, SomeFactoryFnToken, SomeClassToInstantiate } from './index';
 import { TodoAppCmp } from './todo/todo-app.component';
+import { AsyncExampleComponent } from './components/async-example/async-example.component';
 
 @Component( {
   selector: 'my-app',
-  directives: [ TodoAppCmp ],
+  directives: [ TodoAppCmp, AsyncExampleComponent ],
   templateUrl: './app/app.component.html'
 } )
 export class AppComponent implements OnInit {

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

@@ -0,0 +1,84 @@
+import { Observable } from 'rxjs/Observable'
+import { Subscriber } from 'rxjs/Subscriber'
+import 'rxjs/add/observable/interval';
+import 'rxjs/add/operator/take';
+import 'rxjs/add/operator/scan';
+import { Component, Inject } from 'ng-metadata/core';
+
+@Component({
+  selector: 'async-task',
+  template: (
+    `Clock: {{ $ctrl.clock$ | async:this | date:'M/d/yy h:mm:ss a' }} <br>
+     Timer: {{ $ctrl.timer$ | async:this }} <br>
+     Async Stream via scan:
+     <ul><li ng-repeat="u in $ctrl.stream$ | async:this">{{ u }}</li></ul>`
+  )
+})
+class AsyncTaskComponent {
+
+  clock$ = Observable.create((observer: Subscriber<number>) => {
+    this.$interval(() => observer.next(new Date().getTime()), 500);
+  });
+  timer$ = Observable.interval(1000).take(50);
+  stream$ = Observable.interval(1500).take(10).scan((acc, item) => [...acc, item], []);
+
+  constructor(@Inject('$interval') private $interval: ng.IIntervalService){}
+
+}
+
+@Component({
+  selector: 'async-example',
+  directives: [AsyncTaskComponent],
+  template: (
+    `<button ng-click="$ctrl.renderTimers=!$ctrl.renderTimers">Show Timers</button>
+     <async-task ng-if="$ctrl.renderTimers"></async-task>
+     <br>
+    <div>
+      <p>Wait for it... {{ $ctrl.greeting | async }}</p>
+      <button ng-click="$ctrl.clicked()">{{ $ctrl.arrived ? 'Reset' : 'Resolve' }}</button>
+    </div>
+    <div>
+      <h4>Rx repos:</h4>
+      <blockquote ng-hide="$ctrl.repos | async">Loading...</blockquote>
+      <ul>
+        <li ng-repeat="repo in $ctrl.repos | async">
+          <a href="{{ repo.html_url }}" target="_blank">
+            {{ repo.name }}
+          </a>
+        </li>
+      </ul>
+      <pre style="overflow:auto;max-height:250px;">{{ $ctrl.repos | async | json }}</pre>
+    </div>
+    `
+  )
+})
+export class AsyncExampleComponent {
+
+  greeting: ng.IPromise<string> = null;
+  arrived: boolean = false;
+
+  repos = this.$http.get('https://api.github.com/orgs/Reactive-Extensions/repos');
+
+  private resolve: Function = null;
+
+  constructor(
+    @Inject('$q') private $q:ng.IQService,
+    @Inject('$http') private $http:ng.IHttpService
+  ) {
+    this.reset();
+  }
+
+  reset() {
+    this.arrived = false;
+    this.greeting = this.$q((resolve, reject) => { this.resolve = resolve; });
+  }
+
+  clicked() {
+    if (this.arrived) {
+      this.reset();
+    } else {
+      this.resolve('hi there!');
+      this.arrived = true;
+    }
+  }
+}

playground/app/main.ts

@@ -1,11 +1,12 @@
 //main entry point
 import { bootstrap } from 'ng-metadata/platform';
 import { Title } from 'ng-metadata/platform';
-import {enableProdMode} from 'ng-metadata/core';
+import { AsyncPipe } from 'ng-metadata/common';
+import { enableProdMode } from 'ng-metadata/core';
 
 import { AppComponent } from './app.component';
 import { AppModule, configureProviders } from './index';
 
 // enableProdMode();
 
-bootstrap( AppComponent, [ Title, AppModule, configureProviders ] );
+bootstrap( AppComponent, [ Title, AsyncPipe, AppModule, configureProviders ] );

playground/ng-metadata.legacy.d.ts

@@ -1,5 +1,6 @@
 // import Subject = require('rxjs/Subject');
-// import Subscription = require('rxjs/Subscription');
+declare type Observable<T> = any;
+declare type Subject<T> = any;
 
 declare module ngMetadataPlatform{
 
@@ -987,6 +988,18 @@ declare module ngMetadataCommon {
     abstract hasOption(value: any): boolean;
     registerOption(optionScope: ng.IScope, optionElement: ng.IAugmentedJQuery, optionAttrs: ng.IAttributes, interpolateValueFn?: Function, interpolateTextFn?: Function): void;
   }
+
+  export class AsyncPipe {
+    private static nextObjectID;
+    private static values;
+    private static subscriptions;
+    private static TRACK_PROP_NAME;
+    private static objectId(obj);
+    private static isPromiseOrObservable(obj);
+    private static getSubscriptionStrategy(input);
+    private static dispose(inputId);
+    transform(input: Observable<any> | ng.IPromise<any> | ng.IHttpPromise<any>, scope?: ng.IScope): any;
+  }
 }
 declare module "ng-metadata/core" {
   export = ngMetadataCore;

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,113 @@
+import { Subscription } from 'rxjs/Subscription';
+import { Observable } from 'rxjs/Observable';
+
+import { Pipe } from '../../core/pipes/decorators';
+import { PipeTransform } from '../../core/pipes/pipe_interfaces';
+
+import { isObservable, isScope, isSubscription, isPromiseOrObservable } from '../../facade/lang';
+
+type StoredSubscription = ng.IPromise<any>|ng.IHttpPromise<any>|Subscription;
+
+/**
+ * Based on @cvuorinen angular1-async-filter implementation
+ * @link https://github.com/cvuorinen/angular1-async-filter
+ *
+ * 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. ( runs $scope.$digest() )
+ * When the component gets destroyed, the `async` pipe unsubscribes automatically to avoid potential memory leaks.
+ *
+ * ## Usage
+ *
+ *  object | async // for non observable
+ *  object | async:this // for observable
+ *
+ * where:
+ *  - `object` is one of type `Observable`, `Promise`, 'ng.IPromise' or 'ng.IHttpPromise'
+ *  - `this` is pipe parameter ( in angular 1 reference to local $Scope ( we need for Observable disposal )
+ *
+ *  If you are using async with observables nad you don't provide scope we will throw Error to let you know that you forgot `this`, #perfmatters baby!
+ *
+ * ## Examples
+ *
+ * This example binds a `Promise` to the view. Clicking the `Resolve` button resolves the promise.
+ *
+ * {@example core/pipes/ts/async_pipe/async_pipe_example.ts region='AsyncPipePromise'}
+ *
+ * It's also possible to use `async` with Observables. The example below binds the `time` Observable
+ * to the view. Every 500ms, the `time` Observable updates the view with the current time.
+ *
+ * {@example core/pipes/ts/async_pipe/async_pipe_example.ts region='AsyncPipeObservable'}
+ */
+@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 _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 _markForCheck( scope: ng.IScope ) {
+    if ( isScope( scope ) ) {
+      // #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() );
+      // 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 {
+    if ( isSubscription( AsyncPipe.subscriptions[ inputId ] ) ) {
+      (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 ( !isPromiseOrObservable( input ) ) {
+      return input
+    }
+
+    if ( isObservable( input ) && !isScope( scope ) ) {
+      throw new Error( 'AsyncPipe: you have to specify "this" as parameter so we can unsubscribe on scope.$destroy!' );
+    }
+
+    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 ( isScope( scope ) ) {
+      // 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
+      AsyncPipe._markForCheck( scope );
+    }
+
+  }
+}

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

@@ -1,3 +1,6 @@
+import { Subscription } from 'rxjs/Subscription';
+import { Observable } from 'rxjs/Observable';
+
 export interface BrowserNodeGlobal {
   Object: typeof Object,
   Array: typeof Array,
@@ -133,6 +136,26 @@ 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<T>( obj: any ): obj is Observable<T> {
+  return Boolean( isPresent( obj ) && obj.subscribe );
+}
+
+export function isPromiseOrObservable( obj: any ): boolean {
+  return isPromiseLike( obj ) || isObservable( obj );
+}
+
+export function isScope( obj: any ): obj is ng.IScope {
+  return isPresent( obj ) && obj.$digest && obj.$on;
+}
+
+export function isSubscription( obj: any ): obj is Subscription {
+  return isPresent( obj ) && obj.unsubscribe;
+}
+
 export function isJsObject( o: any ): boolean {
   return o !== null && (typeof o === "function" || typeof o === "object");
 }