feat(throwError): functional version of throwError

- eliminates `ErrorObservable`
- eliminates `_throw`
- updates tests

BREAKING CHANGE: Observable.throw no longer available in TypeScript without a cast

Author: benlesh

spec/exports-spec.ts

@@ -19,7 +19,7 @@ import { onErrorResumeNext } from '../src/internal/observable/onErrorResumeNext'
 import { pairs } from '../src/internal/observable/pairs';
 import { race } from '../src/internal/observable/race';
 import { range } from '../src/internal/observable/range';
-import { _throw } from '../src/internal/observable/throw';
+import { throwError } from '../src/internal/observable/throwError';
 import { timer } from '../src/internal/observable/timer';
 import { using } from '../src/internal/observable/using';
 import { zip } from '../src/internal/observable/zip';
@@ -106,8 +106,8 @@ describe('exports', () => {
     expect(range).to.equal(Rx.Observable.range);
   });
 
-  it('should have rxjs/observable/throw', () => {
-    expect(_throw).to.equal(Rx.Observable.throw);
+  it('should have rxjs/observable/throwError', () => {
+    expect(throwError).to.equal(Rx.Observable.throw);
   });
 
   it('should have rxjs/observable/timer', () => {

spec/observables/throw-spec.ts renamed to spec/observables/throwError-spec.ts

@@ -1,24 +1,23 @@
 import { expect } from 'chai';
 import * as Rx from '../../src/Rx';
-import { ErrorObservable } from '../../src/internal/observable/ErrorObservable';
+import { throwError } from '../../src/create';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
 
-declare const { asDiagram };
+declare const asDiagram: any;
 declare const expectObservable: typeof marbleTestingSignature.expectObservable;
 declare const rxTestScheduler: Rx.TestScheduler;
-const Observable = Rx.Observable;
 
 /** @test {throw} */
-describe('Observable.throw', () => {
+describe('throwError', () => {
   asDiagram('throw(e)')('should create a cold observable that just emits an error', () => {
     const expected = '#';
-    const e1 = Observable.throw('error');
+    const e1 = throwError('error');
     expectObservable(e1).toBe(expected);
   });
 
   it('should emit one value', (done: MochaDone) => {
     let calls = 0;
-    Observable.throw('bad').subscribe(() => {
+    throwError('bad').subscribe(() => {
       done(new Error('should not be called'));
     }, (err: any) => {
       expect(++calls).to.equal(1);
@@ -27,21 +26,8 @@ describe('Observable.throw', () => {
     });
   });
 
-  it('should create expose a error property', () => {
-    const e = Observable.throw('error');
-
-    expect(e['error']).to.equal('error');
-  });
-
-  it('should create ErrorObservable via static create function', () => {
-    const e = new ErrorObservable('error');
-    const r = ErrorObservable.create('error');
-
-    expect(e).to.deep.equal(r);
-  });
-
   it('should accept scheduler', () => {
-    const e = Observable.throw('error', rxTestScheduler);
+    const e = throwError('error', rxTestScheduler);
 
     expectObservable(e).toBe('#');
   });

src/add/observable/throw.ts

@@ -1,4 +1,4 @@
 import { Observable } from '../../internal/Observable';
-import { _throw } from '../../internal/observable/throw';
+import { throwError } from '../../internal/observable/throwError';
 
-Observable.throw = _throw;
+(Observable as any).throw = throwError;

src/create/index.ts

@@ -19,7 +19,7 @@ export { onErrorResumeNext } from '../internal/observable/onErrorResumeNext';
 export { pairs } from '../internal/observable/pairs';
 export { race } from '../internal/observable/race';
 export { range } from '../internal/observable/range';
-export { _throw as throwError } from '../internal/observable/throw';
+export { throwError } from '../internal/observable/throwError';
 export { timer } from '../internal/observable/timer';
 export { using } from '../internal/observable/using';
 export { zip } from '../internal/observable/zip';

src/internal/Notification.ts

@@ -2,7 +2,7 @@ import { PartialObserver } from './Observer';
 import { Observable } from './Observable';
 import { empty } from './observable/empty';
 import { of } from './observable/of';
-import { _throw } from './observable/throw';
+import { throwError } from './observable/throwError';
 
 /**
  * Represents a push-based event or value that an {@link Observable} can emit.
@@ -89,7 +89,7 @@ export class Notification<T> {
       case 'N':
         return of(this.value);
       case 'E':
-        return _throw(this.error);
+        return throwError(this.error);
       case 'C':
         return empty();
     }

src/internal/Observable.ts

@@ -5,7 +5,6 @@ import { Subscription, AnonymousSubscription, TeardownLogic } from './Subscripti
 import { root } from './util/root';
 import { toSubscriber } from './util/toSubscriber';
 import { IfObservable } from './observable/IfObservable';
-import { ErrorObservable } from './observable/ErrorObservable';
 import { observable as Symbol_observable } from '../internal/symbol/observable';
 import { OperatorFunction } from '../internal/types';
 import { pipeFromArray } from './util/pipe';
@@ -256,7 +255,6 @@ export class Observable<T> implements Subscribable<T> {
 
   // `if` and `throw` are special snow flakes, the compiler sees them as reserved words
   static if: typeof IfObservable.create;
-  static throw: typeof ErrorObservable.create;
 
   /**
    * An interop point defined by the es7-observable spec https://github.com/zenparsing/es-observable

src/internal/observable/ErrorObservable.ts

@@ -0,0 +1,60 @@
+import { Observable } from '../Observable';
+import { IScheduler } from '../Scheduler';
+import { Subscriber } from '../Subscriber';
+
+/**
+ * Creates an Observable that emits no items to the Observer and immediately
+ * emits an error notification.
+ *
+ * <span class="informal">Just emits 'error', and nothing else.
+ * </span>
+ *
+ * <img src="./img/throw.png" width="100%">
+ *
+ * This static operator is useful for creating a simple Observable that only
+ * emits the error notification. It can be used for composing with other
+ * Observables, such as in a {@link mergeMap}.
+ *
+ * @example <caption>Emit the number 7, then emit an error.</caption>
+ * var result = Rx.Observable.throw(new Error('oops!')).startWith(7);
+ * result.subscribe(x => console.log(x), e => console.error(e));
+ *
+ * @example <caption>Map and flatten numbers to the sequence 'a', 'b', 'c', but throw an error for 13</caption>
+ * var interval = Rx.Observable.interval(1000);
+ * var result = interval.mergeMap(x =>
+ *   x === 13 ?
+ *     Rx.Observable.throw('Thirteens are bad') :
+ *     Rx.Observable.of('a', 'b', 'c')
+ * );
+ * result.subscribe(x => console.log(x), e => console.error(e));
+ *
+ * @see {@link create}
+ * @see {@link empty}
+ * @see {@link never}
+ * @see {@link of}
+ *
+ * @param {any} error The particular Error to pass to the error notification.
+ * @param {Scheduler} [scheduler] A {@link IScheduler} to use for scheduling
+ * the emission of the error notification.
+ * @return {Observable} An error Observable: emits only the error notification
+ * using the given error argument.
+ * @static true
+ * @name throw
+ * @owner Observable
+ */
+export function throwError(error: any, scheduler?: IScheduler): Observable<never> {
+  if (!scheduler) {
+    return new Observable(subscriber => subscriber.error(error));
+  } else {
+    return new Observable(subscriber => scheduler.schedule(dispatch, 0, { error, subscriber }));
+  }
+}
+
+interface DispatchArg {
+  error: any;
+  subscriber: Subscriber<any>;
+}
+
+function dispatch({ error, subscriber }: DispatchArg) {
+  subscriber.error(error);
+}