feat(empty): empty() returns the same instance

- Adds more robust tests
- Ensures that all calls to empty without a scheduler return a single instance
- Removes EmptyObservable
- Updates other tests and code using EmptyObservable
EmptyObservable
BREAKING CHANGE: `empty()` without a scheduler will return the same
instance every time.
BREAKING CHANGE: In TypeScript, `empty()` no longer accepts a generic
argument, as it returns `Observable<never>`

spec/observables/empty-spec.ts

@@ -1,16 +1,42 @@
-import * as Rx from '../../src/Rx';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
+import { empty } from '../../src/create';
+import { expect } from 'chai';
 
-declare const { asDiagram };
+declare const asDiagram: any;
 declare const expectObservable: typeof marbleTestingSignature.expectObservable;
-
-const Observable = Rx.Observable;
+declare const rxTestScheduler: any;
 
 /** @test {empty} */
-describe('Observable.empty', () => {
+describe('empty', () => {
   asDiagram('empty')('should create a cold observable with only complete', () => {
     const expected = '|';
-    const e1 = Observable.empty();
+    const e1 = empty();
     expectObservable(e1).toBe(expected);
   });
+
+  it('should return the same instance EMPTY', () => {
+    const s1 = empty();
+    const s2 = empty();
+    expect(s1).to.equal(s2);
+  });
+
+  it('should be synchronous by default', () => {
+    const source = empty();
+    let hit = false;
+    source.subscribe({
+      complete() { hit = true; }
+    });
+    expect(hit).to.be.true;
+  });
+
+  it('should take a scheduler', () => {
+    const source = empty(rxTestScheduler);
+    let hit = false;
+    source.subscribe({
+      complete() { hit = true; }
+    });
+    expect(hit).to.be.false;
+    rxTestScheduler.flush();
+    expect(hit).to.be.true;
+  });
 });

spec/observables/of-spec.ts

@@ -1,9 +1,8 @@
 import { expect } from 'chai';
 import * as Rx from '../../src/Rx';
-import { EmptyObservable } from '../../src/internal/observable/EmptyObservable';
+import { empty } from '../../src/internal/observable/empty';
 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;
@@ -35,12 +34,7 @@ describe('Observable.of', () => {
 
   it('should return an empty observable if passed no values', () => {
     const obs = Observable.of();
-    expect(obs instanceof EmptyObservable).to.be.true;
-  });
-
-  it('should return an empty observable if passed only a scheduler', () => {
-    const obs = Observable.of(Rx.Scheduler.queue);
-    expect(obs instanceof EmptyObservable).to.be.true;
+    expect(obs).to.equal(empty());
   });
 
   it('should emit one value', (done: MochaDone) => {

src/internal/Notification.ts

@@ -91,7 +91,7 @@ export class Notification<T> {
       case 'E':
         return _throw(this.error);
       case 'C':
-        return empty<T>();
+        return empty();
     }
     throw new Error('unexpected notification kind value');
   }

src/internal/observable/ForkJoinObservable.ts

@@ -1,7 +1,7 @@
 import { Observable, SubscribableOrPromise } from '../Observable';
 import { Subscriber } from '../Subscriber';
 import { Subscription } from '../Subscription';
-import { EmptyObservable } from './EmptyObservable';
+import { empty } from './empty';
 import { isArray } from '..//util/isArray';
 
 import { subscribeToResult } from '..//util/subscribeToResult';
@@ -140,7 +140,7 @@ export class ForkJoinObservable<T> extends Observable<T> {
                                   Array<SubscribableOrPromise<any>> |
                                   ((...values: Array<any>) => any)>): Observable<T> {
     if (sources === null || arguments.length === 0) {
-      return new EmptyObservable<T>();
+      return empty();
     }
 
     let resultSelector: (...values: Array<any>) => any = null;
@@ -155,7 +155,7 @@ export class ForkJoinObservable<T> extends Observable<T> {
     }
 
     if (sources.length === 0) {
-      return new EmptyObservable<T>();
+      return empty();
     }
 
     return new ForkJoinObservable(<Array<SubscribableOrPromise<any>>>sources, resultSelector);

src/internal/observable/empty.ts

@@ -1,3 +1,55 @@
-import {  EmptyObservable  } from './EmptyObservable';
+import { Observable } from '../Observable';
+import { IScheduler } from '../Scheduler';
 
-export const empty = EmptyObservable.create;
+export const EMPTY = new Observable<never>(subscriber => subscriber.complete());
+
+/**
+ * Creates an Observable that emits no items to the Observer and immediately
+ * emits a complete notification.
+ *
+ * <span class="informal">Just emits 'complete', and nothing else.
+ * </span>
+ *
+ * <img src="./img/empty.png" width="100%">
+ *
+ * This static operator is useful for creating a simple Observable that only
+ * emits the complete notification. It can be used for composing with other
+ * Observables, such as in a {@link mergeMap}.
+ *
+ * @example <caption>Emit the number 7, then complete.</caption>
+ * var result = Rx.Observable.empty().startWith(7);
+ * result.subscribe(x => console.log(x));
+ *
+ * @example <caption>Map and flatten only odd numbers to the sequence 'a', 'b', 'c'</caption>
+ * var interval = Rx.Observable.interval(1000);
+ * var result = interval.mergeMap(x =>
+ *   x % 2 === 1 ? Rx.Observable.of('a', 'b', 'c') : Rx.Observable.empty()
+ * );
+ * result.subscribe(x => console.log(x));
+ *
+ * // Results in the following to the console:
+ * // x is equal to the count on the interval eg(0,1,2,3,...)
+ * // x will occur every 1000ms
+ * // if x % 2 is equal to 1 print abc
+ * // if x % 2 is not equal to 1 nothing will be output
+ *
+ * @see {@link create}
+ * @see {@link never}
+ * @see {@link of}
+ * @see {@link throw}
+ *
+ * @param {Scheduler} [scheduler] A {@link IScheduler} to use for scheduling
+ * the emission of the complete notification.
+ * @return {Observable} An "empty" Observable: emits only the complete
+ * notification.
+ * @static true
+ * @name empty
+ * @owner Observable
+ */
+export function empty(scheduler?: IScheduler) {
+  return scheduler ? emptyScheduled(scheduler) : EMPTY;
+}
+
+export function emptyScheduled(scheduler: IScheduler) {
+  return new Observable<never>(subscriber => scheduler.schedule(() => subscriber.complete()));
+}

src/internal/operators/repeat.ts

@@ -1,7 +1,7 @@
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { Observable } from '../Observable';
-import { EmptyObservable } from '../observable/EmptyObservable';
+import { empty } from '../observable/empty';
 import { TeardownLogic } from '../Subscription';
 import { MonoTypeOperatorFunction } from '../../internal/types';
 
@@ -20,7 +20,7 @@ import { MonoTypeOperatorFunction } from '../../internal/types';
 export function repeat<T>(count: number = -1): MonoTypeOperatorFunction<T> {
   return (source: Observable<T>) => {
     if (count === 0) {
-      return new EmptyObservable<T>();
+      return empty();
     } else if (count < 0) {
       return source.lift(new RepeatOperator(-1, source));
     } else {

src/internal/operators/startWith.ts

@@ -2,7 +2,7 @@ import { IScheduler } from '../Scheduler';
 import { Observable } from '../Observable';
 import { fromArray } from '../observable/fromArray';
 import { scalar } from '../observable/scalar';
-import { EmptyObservable } from '../observable/EmptyObservable';
+import { empty } from '../observable/empty';
 import { concat as concatStatic } from '../observable/concat';
 import { isScheduler } from '..//util/isScheduler';
 import { MonoTypeOperatorFunction } from '../../internal/types';
@@ -46,7 +46,7 @@ export function startWith<T>(...array: Array<T | IScheduler>): MonoTypeOperatorF
     } else if (len > 0) {
       return concatStatic(fromArray(array as T[], scheduler), source);
     } else {
-      return concatStatic(new EmptyObservable<T>(scheduler), source);
+      return concatStatic<T>(empty(scheduler), source);
     }
   };
 }

src/internal/operators/take.ts

@@ -1,7 +1,7 @@
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { ArgumentOutOfRangeError } from '..//util/ArgumentOutOfRangeError';
-import { EmptyObservable } from '../observable/EmptyObservable';
+import { empty } from '../observable/empty';
 import { Observable } from '../Observable';
 import { TeardownLogic } from '../Subscription';
 import { MonoTypeOperatorFunction } from '../../internal/types';
@@ -42,7 +42,7 @@ import { MonoTypeOperatorFunction } from '../../internal/types';
 export function take<T>(count: number): MonoTypeOperatorFunction<T> {
   return (source: Observable<T>) => {
     if (count === 0) {
-      return new EmptyObservable<T>();
+      return empty();
     } else {
       return source.lift(new TakeOperator(count));
     }

src/internal/operators/takeLast.ts

@@ -1,7 +1,7 @@
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { ArgumentOutOfRangeError } from '..//util/ArgumentOutOfRangeError';
-import { EmptyObservable } from '../observable/EmptyObservable';
+import { empty } from '../observable/empty';
 import { Observable } from '../Observable';
 import { TeardownLogic } from '../Subscription';
 import { MonoTypeOperatorFunction } from '../../internal/types';
@@ -45,7 +45,7 @@ import { MonoTypeOperatorFunction } from '../../internal/types';
 export function takeLast<T>(count: number): MonoTypeOperatorFunction<T> {
   return function takeLastOperatorFunction(source: Observable<T>): Observable<T> {
     if (count === 0) {
-      return new EmptyObservable<T>();
+      return empty();
     } else {
       return source.lift(new TakeLastOperator(count));
     }