fix(catch): accept selector returns ObservableInput

closes #1857

spec/helpers/test-helper.ts

@@ -3,7 +3,10 @@ declare const global: any;
 declare const Symbol: any;
 
 import * as Rx from '../../dist/cjs/Rx';
+import {ObservableInput} from '../../dist/cjs/Observable';
 import {root} from '../../dist/cjs/util/root';
+import {$$iterator} from '../../dist/cjs/symbol/iterator';
+import $$symbolObservable from 'symbol-observable';
 
 export function lowerCaseO<T>(...args): Rx.Observable<T> {
   const values = [].slice.apply(arguments);
@@ -24,4 +27,19 @@ export function lowerCaseO<T>(...args): Rx.Observable<T> {
   return <any>o;
 };
 
+export const createObservableInputs = <T>(value: T) => Rx.Observable.of<ObservableInput<T>>(
+  Rx.Observable.of<T>(value),
+  Rx.Observable.of<T>(value, Rx.Scheduler.async),
+  [value],
+  Promise.resolve(value),
+  <any>({ [$$iterator]: () => {
+      return {
+        next: () => {
+          return value;
+        }
+      };
+    }}),
+  <any>({ [$$symbolObservable]: () => Rx.Observable.of(value) })
+);
+
 global.__root__ = root;

spec/operators/catch-spec.ts

@@ -1,5 +1,6 @@
 import {expect} from 'chai';
 import * as Rx from '../../dist/cjs/Rx';
+import {createObservableInputs} from '../helpers/test-helper';
 declare const {hot, cold, asDiagram, expectObservable, expectSubscriptions};
 
 declare const rxTestSchdeuler: Rx.TestScheduler;
@@ -227,4 +228,18 @@ describe('Observable.prototype.catch', () => {
           done();
         });
   });
+
+  it('should accept selector returns any ObservableInput', (done: MochaDone) => {
+    const input$ = createObservableInputs(42);
+
+    input$.mergeMap(input =>
+      Observable.throw('bad').catch(err => input)
+    ).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);
-  }
 }