fix(catch): accept selector returns ObservableInput

closes ReactiveX#1857

spec/operators/catch-spec.ts

@@ -227,4 +227,16 @@ describe('Observable.prototype.catch', () => {
           done();
         });
   });
+
+  it('should accept selector returns promise', (done: MochaDone) => {
+    Observable.throw('bad')
+      .catch(err => Promise.resolve(42))
+      .subscribe(x => {
+        expect(x).to.be.equal(42);
+       }, (err: any) => {
+          done(new Error('should not be called'));
+        }, () => {
+          done();
+        });
+  });
 });

src/operator/catch.ts

@@ -1,6 +1,9 @@
 import {Operator} from '../Operator';
 import {Subscriber} from '../Subscriber';
-import {Observable} from '../Observable';
+import {Observable, ObservableInput} from '../Observable';
+
+import {OuterSubscriber} from '../OuterSubscriber';
+import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
  * Catches errors on the observable to be handled by returning a new observable or throwing an error.
@@ -12,21 +15,21 @@ import {Observable} from '../Observable';
  * @method catch
  * @owner Observable
  */
-export function _catch<T, R>(selector: (err: any, caught: Observable<T>) => Observable<R>): Observable<R> {
+export function _catch<T, R>(selector: (err: any, caught: Observable<T>) => ObservableInput<R>): Observable<R> {
   const operator = new CatchOperator(selector);
   const caught = this.lift(operator);
   return (operator.caught = caught);
 }
 
 export interface CatchSignature<T> {
-  (selector: (err: any, caught: Observable<T>) => Observable<T>): Observable<T>;
-  <R>(selector: (err: any, caught: Observable<T>) => Observable<R>): Observable<R>;
+  (selector: (err: any, caught: Observable<T>) => ObservableInput<T>): Observable<T>;
+  <R>(selector: (err: any, caught: Observable<T>) => ObservableInput<R>): Observable<R>;
 }
 
 class CatchOperator<T, R> implements Operator<T, R> {
-  caught: Observable<any>;
+  caught: Observable<T>;
 
-  constructor(private selector: (err: any, caught: Observable<any>) => Observable<any>) {
+  constructor(private selector: (err: any, caught: Observable<T>) => ObservableInput<T | R>) {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
@@ -39,11 +42,10 @@ class CatchOperator<T, R> implements Operator<T, R> {
  * @ignore
  * @extends {Ignored}
  */
-class CatchSubscriber<T> extends Subscriber<T> {
-
+class CatchSubscriber<T, R> extends OuterSubscriber<T, R> {
   constructor(destination: Subscriber<any>,
-              private selector: (err: any, caught: Observable<any>) => Observable<any>,
-              private caught: Observable<any>) {
+              private selector: (err: any, caught: Observable<T>) => ObservableInput<T | R>,
+              private caught: Observable<T>) {
     super(destination);
   }
 
@@ -60,13 +62,9 @@ class CatchSubscriber<T> extends Subscriber<T> {
         return;
       }
 
-      this._innerSub(result);
+      this.unsubscribe();
+      (<any>this.destination).remove(this);
+      subscribeToResult(this, result);
     }
   }
-
-  private _innerSub(result: Observable<any>) {
-    this.unsubscribe();
-    (<any>this.destination).remove(this);
-    result.subscribe(this.destination);
-  }
 }