Merge cf53d29 into 91d019c

spec/observables/defer-spec.ts

@@ -1,5 +1,6 @@
 import * as Rx from '../../dist/cjs/Rx';
-declare const {hot, expectObservable, expectSubscriptions};
+import {DoneSignature} from '../helpers/test-helper';
+declare const {hot, expectObservable, expectSubscriptions, type};
 
 const Observable = Rx.Observable;
 
@@ -27,6 +28,36 @@ describe('Observable.defer', () => {
     expectSubscriptions(source.subscriptions).toBe(sourceSubs);
   });
 
+  it('should accept factory returns promise resolves', (done: DoneSignature) => {
+    const expected = 42;
+    const e1 = Observable.defer(() => {
+      return new Promise((resolve: any) => { resolve(expected); });
+    });
+
+    e1.subscribe((x: number) => {
+      expect(x).toBe(expected);
+      done();
+    }, x => {
+      done.fail();
+    });
+  });
+
+  it('should accept factory returns promise rejects', (done: DoneSignature) => {
+    const expected = 42;
+    const e1 = Observable.defer(() => {
+      return new Promise((resolve: any, reject: any) => { reject(expected); });
+    });
+
+    e1.subscribe((x: number) => {
+      done.fail();
+    }, x => {
+      expect(x).toBe(expected);
+      done();
+    }, () => {
+      done.fail();
+    });
+  });
+
   it('should create an observable from error', () => {
     const source = hot('#');
     const sourceSubs = '(^!)';
@@ -39,10 +70,9 @@ describe('Observable.defer', () => {
   });
 
   it('should create an observable when factory throws', () => {
-    //type definition need to be updated
-    const e1 = Observable.defer(<any>(() => {
+    const e1 = Observable.defer(() => {
       throw 'error';
-    }));
+    });
     const expected = '#';
 
     expectObservable(e1).toBe(expected);

src/observable/DeferObservable.ts

@@ -1,8 +1,9 @@
-import {Observable} from '../Observable';
+import {Observable, SubscribableOrPromise} from '../Observable';
 import {Subscriber} from '../Subscriber';
-import {tryCatch} from '../util/tryCatch';
-import {errorObject} from '../util/errorObject';
+import {Subscription} from '../Subscription';
 
+import {subscribeToResult} from '../util/subscribeToResult';
+import {OuterSubscriber} from '../OuterSubscriber';
 /**
  *
  */
@@ -15,20 +16,34 @@ export class DeferObservable<T> extends Observable<T> {
    * @name defer
    * @owner Observable
    */
-  static create<T>(observableFactory: () => Observable<T>): Observable<T> {
+  static create<T>(observableFactory: () => SubscribableOrPromise<T> | void): Observable<T> {
     return new DeferObservable(observableFactory);
   }
 
-  constructor(private observableFactory: () => Observable<T>) {
+  constructor(private observableFactory: () => SubscribableOrPromise<T> | void) {
     super();
   }
 
-  protected _subscribe(subscriber: Subscriber<T>) {
-    const result = tryCatch(this.observableFactory)();
-    if (result === errorObject) {
-      subscriber.error(errorObject.e);
-    } else {
-      result.subscribe(subscriber);
+  protected _subscribe(subscriber: Subscriber<T>): Subscription {
+    return new DeferSubscriber(subscriber, this.observableFactory);
+  }
+}
+
+class DeferSubscriber<T> extends OuterSubscriber<T, T> {
+  constructor(destination: Subscriber<T>,
+              private factory: () => SubscribableOrPromise<T> | void) {
+    super(destination);
+    this.tryDefer();
+  }
+
+  private tryDefer(): void {
+    try {
+      const result = this.factory.call(this);
+      if (result) {
+        this.add(subscribeToResult(this, result));
+      }
+    } catch (err) {
+      this._error(err);
     }
   }
 }