feat(core): added *async directive

Example usages:
<ng-container *async="let data from async; next next; error error; complete complete">{{ data }}</ng-container>
<ng-template [async]="async" (next)="next($event)" (error)="error($event)" (complete)="complete($event)" let-data>{{ data }}</ng-template>

projects/platform/src/lib/directives/async.directive.ts

@@ -0,0 +1,149 @@
+import { Directive, EmbeddedViewRef, EventEmitter, Input, OnChanges, OnDestroy, Output, SimpleChanges, TemplateRef, ViewContainerRef } from '@angular/core';
+import { isObservable, Observable, SubscriptionLike } from 'rxjs';
+
+type ObservableOrPromise<T> = Observable<T> | Promise<T>;
+
+interface AsyncContext {
+  $implicit: any;
+}
+
+interface SubscriptionStrategy {
+  createSubscription(async: ObservableOrPromise<any>, next: any, error: any, complete: any): SubscriptionLike | Promise<any>;
+
+  dispose(subscription: SubscriptionLike | Promise<any>): void;
+}
+
+class ObservableStrategy implements SubscriptionStrategy {
+  createSubscription(async: ObservableOrPromise<any>, next: any, error: any, complete: any): SubscriptionLike {
+    return async.subscribe(next, error, complete);
+  }
+
+  dispose(subscription: SubscriptionLike): void {
+    if (subscription) {
+      subscription.unsubscribe();
+    }
+  }
+}
+
+class PromiseStrategy implements SubscriptionStrategy {
+  createSubscription(async: Observable<any> | Promise<any>, next: any, error: any, complete: any): Promise<any> {
+    return async.then(next, error).finally(complete);
+  }
+
+  dispose(subscription: Promise<any>): void {}
+}
+
+const observableStrategy = new ObservableStrategy();
+const promiseStrategy = new PromiseStrategy();
+
+function resolveStrategy(async: ObservableOrPromise<any>): SubscriptionStrategy {
+  if (isObservable(async)) {
+    return observableStrategy;
+  }
+
+  if (isPromise(async)) {
+    return promiseStrategy;
+  }
+
+  throw new Error(`InvalidDirectiveArgument: 'async' for directive 'async'`);
+}
+
+@Directive({ selector: '[async]' })
+export class AsyncDirective implements OnChanges, OnDestroy {
+
+  @Input() async: ObservableOrPromise<any>;
+  @Input() asyncFrom: ObservableOrPromise<any>;
+  @Input() asyncNext: any;
+  @Input() asyncError: any;
+  @Input() asyncComplete: any;
+
+  @Output() next: EventEmitter<any> = new EventEmitter<any>();
+  @Output() error: EventEmitter<any> = new EventEmitter<any>();
+  @Output() complete: EventEmitter<any> = new EventEmitter<any>();
+
+  private context: AsyncContext = { $implicit: null };
+  private viewRef: EmbeddedViewRef<AsyncContext> =
+    this.viewContainer.createEmbeddedView(this.templateRef, this.context);
+
+  private strategy: SubscriptionStrategy;
+  private subscription: SubscriptionLike | Promise<any>;
+
+  constructor(
+    private templateRef: TemplateRef<AsyncContext>,
+    private viewContainer: ViewContainerRef
+  ) {}
+
+  ngOnChanges(changes: SimpleChanges) {
+    if ('async' in changes) {
+      this.onAsyncDidChanged(this.async, changes.async.previousValue);
+    }
+    if ('asyncFrom' in changes) {
+      this.onAsyncDidChanged(this.asyncFrom, changes.asyncFrom.previousValue);
+    }
+  }
+
+  ngOnDestroy() {
+    this.dispose();
+  }
+
+  private onAsyncDidChanged(current: ObservableOrPromise<any>, previous: ObservableOrPromise<any>): void {
+    if (!this.subscription) {
+      return current && this.subscribe(current);
+    }
+
+    if (current !== previous) {
+      this.dispose();
+      return this.onAsyncDidChanged(current, null);
+    }
+  }
+
+  private subscribe(async: ObservableOrPromise<async>) {
+    this.strategy = resolveStrategy(async);
+    this.subscription = this.strategy.createSubscription(
+      async,
+      (value: any) => this.onNext(value),
+      (value: any) => this.onError(value),
+      (value: any) => this.onComplete(value)
+    );
+  }
+
+  private onNext(value: any): void {
+    this.context.$implicit = value;
+    this.next.emit(value);
+    if (isFunction(this.asyncNext)) {
+      this.asyncNext(value);
+    }
+    this.viewRef.markForCheck();
+  }
+
+  private onError(value: any): void {
+    this.error.emit(value);
+    if (isFunction(this.asyncError)) {
+      this.asyncError(value);
+    }
+  }
+
+  private onComplete(value: any): void {
+    this.complete.next(value);
+    if (isFunction(this.asyncComplete)) {
+      this.asyncComplete(value);
+    }
+  }
+
+  private dispose(): void {
+    if (this.strategy) {
+      this.strategy.dispose(this.subscription);
+      this.subscription = null;
+      this.strategy = null;
+    }
+  }
+
+}
+
+function isFunction(value: any): value is Function {
+  return typeof value === 'function';
+}
+
+function isPromise<T>(value: any): value is Promise<T> {
+  return value && typeof value.subscribe !== 'function' && typeof value.then === 'function';
+}

projects/platform/src/lib/directives/index.ts

@@ -1,23 +1,26 @@
+import { AsyncDirective } from './async.directive';
+import { ComposeDirective, ReturnDirective } from './compose.directive';
+import { CookiesDirective } from './cookies.directive';
 import { HttpDirective } from './http.directive';
-import { RouteDirective } from './route.directive';
 import { InitDirective } from './init.directive';
+import { RouteDirective } from './route.directive';
 import { TimeoutDirective } from './timeout.directive';
-import { ComposeDirective, ReturnDirective } from './compose.directive';
-import { CookiesDirective } from './cookies.directive';
 
 export const DIRECTIVES = [
-    HttpDirective,
-    RouteDirective,
-    InitDirective,
-    TimeoutDirective,
-    ComposeDirective,
-    ReturnDirective,
-    CookiesDirective
+  AsyncDirective,
+  ComposeDirective,
+  CookiesDirective,
+  HttpDirective,
+  InitDirective,
+  ReturnDirective,
+  RouteDirective,
+  TimeoutDirective
 ];
 
+export * from './async.directive';
+export * from './compose.directive';
+export * from './cookies.directive';
 export * from './http.directive';
-export * from './route.directive';
 export * from './init.directive';
+export * from './route.directive';
 export * from './timeout.directive';
-export * from './compose.directive';
-export * from './cookies.directive';

projects/platform/src/lib/directives/init.directive.ts

@@ -1,24 +1,24 @@
-import { Directive, Input, TemplateRef, ViewContainerRef, EmbeddedViewRef } from '@angular/core';
+import { Directive, EmbeddedViewRef, Input, TemplateRef, ViewContainerRef } from '@angular/core';
 
 interface InitContext {
-    $implicit: any;
+  $implicit: any;
 }
 
-@Directive({selector: '[init]'})
+@Directive({ selector: '[init]' })
 export class InitDirective {
 
-    @Input() set initOf(value: any) {
-        this.context.$implicit = value;
-        this.viewRef.markForCheck();
-    }
+  @Input() set initOf(value: any) {
+    this.context.$implicit = value;
+    this.viewRef.markForCheck();
+  }
 
-    private context: InitContext = { $implicit: null };
-    private viewRef: EmbeddedViewRef<InitContext> =
-        this.viewContainer.createEmbeddedView(this.templateRef, this.context);
+  private context: InitContext = { $implicit: null };
+  private viewRef: EmbeddedViewRef<InitContext> =
+    this.viewContainer.createEmbeddedView(this.templateRef, this.context);
 
-    constructor(
-        private templateRef: TemplateRef<InitContext>,
-        private viewContainer: ViewContainerRef
-    ) { }
+  constructor(
+    private templateRef: TemplateRef<InitContext>,
+    private viewContainer: ViewContainerRef
+  ) { }
 
 }

projects/platform/src/test/directives/async.directive.spec.ts

@@ -0,0 +1,100 @@
+import { Component } from '@angular/core';
+import { fakeAsync, tick } from '@angular/core/testing';
+import { createHostComponentFactory, SpectatorWithHost } from '@netbasal/spectator';
+import { from, Observable, of, throwError } from 'rxjs';
+import { AsyncDirective } from '../../lib/directives';
+
+@Component({ selector: 'host', template: '' })
+class Host {
+  async: Observable<any> | Promise<any>;
+
+  next($event) {}
+
+  error($event) {}
+
+  complete($event) {}
+}
+
+const ERROR_VALUE = 'Async Error';
+const NEXT_VALUE = 'Async Data';
+
+const structuralTemplate = `<ng-container *async="let data from async; next next; error error; complete complete">{{ data }}</ng-container>`;
+const bindingTemplate = `<ng-template [async]="async" (next)="next($event)" (error)="error($event)" (complete)="complete($event)" let-data>{{ data }}</ng-template>`;
+
+describe('AsyncDirective', () => {
+  let host: SpectatorWithHost<AsyncDirective, Host>;
+  const create = createHostComponentFactory({
+    component: AsyncDirective,
+    host: Host
+  });
+
+  [
+    {
+      name: 'structural template',
+      template: structuralTemplate
+    },
+    {
+      name: 'binding template',
+      template: bindingTemplate
+    }
+  ].forEach(({ name, template }) => {
+    describe(`with ${name}`, () => {
+
+      it('should subscribe to observable', () => {
+        host = create(template);
+        spyHost(host.hostComponent);
+        host.setHostInput({ async: from([ 1, 2 ]) });
+
+        expect(host.hostComponent.next).toHaveBeenCalledWith(1);
+        expect(host.hostComponent.next).toHaveBeenCalledWith(2);
+        expect(host.hostComponent.next).toHaveBeenCalledTimes(2);
+        expect(host.hostComponent.error).not.toHaveBeenCalled();
+        expect(host.hostComponent.complete).toHaveBeenCalledTimes(1);
+      });
+
+      it('should subscribe to observable and throw error', () => {
+        host = create(template);
+        spyHost(host.hostComponent);
+        host.setHostInput({ async: throwError(ERROR_VALUE) });
+
+        expect(host.hostComponent.error).toHaveBeenCalledWith(ERROR_VALUE);
+      });
+
+      it('should subscribe to promise', fakeAsync(() => {
+        host = create(template);
+        spyHost(host.hostComponent);
+        host.setHostInput({ async: Promise.resolve(1) });
+        tick();
+
+        expect(host.hostComponent.next).toHaveBeenCalledWith(1);
+        expect(host.hostComponent.next).toHaveBeenCalledTimes(1);
+        expect(host.hostComponent.error).not.toHaveBeenCalled();
+        expect(host.hostComponent.complete).toHaveBeenCalledTimes(1);
+      }));
+
+      it('should subscribe to promise and throw error', fakeAsync(() => {
+        host = create(template);
+        spyHost(host.hostComponent);
+        host.setHostInput({ async: Promise.reject(ERROR_VALUE) });
+        tick();
+
+        expect(host.hostComponent.error).toHaveBeenCalledWith(ERROR_VALUE);
+      }));
+
+      it('should render $implicit context', () => {
+        host = create(template);
+        spyHost(host.hostComponent);
+        host.setHostInput({ async: of(NEXT_VALUE) });
+
+        expect(host.hostElement).toHaveText(NEXT_VALUE);
+      });
+    });
+  });
+
+});
+
+function spyHost(host: Host) {
+  spyOn(host, 'next').and.callThrough();
+  spyOn(host, 'error').and.callThrough();
+  spyOn(host, 'complete').and.callThrough();
+}