feat(operator): add isEmpty operator

perf/micro/immediate-scheduler/operators/isempty.js

@@ -0,0 +1,20 @@
+var RxOld = require("rx");
+var RxNew = require("../../../../index");
+
+module.exports = function (suite) {
+
+    var oldIsEmptyNoArgs = RxOld.Observable.of(25, RxOld.Scheduler.immediate).isEmpty();
+    var newIsEmptyNoArgs = RxNew.Observable.of(25).isEmpty();
+
+    return suite
+        .add('old isEmpty with immediate scheduler', function () {
+            oldIsEmptyNoArgs.subscribe(_next, _error, _complete);
+        })
+        .add('new isEmpty with immediate scheduler', function () {
+            newIsEmptyNoArgs.subscribe(_next, _error, _complete);
+        });
+
+    function _next(x) { }
+    function _error(e){ }
+    function _complete(){ }
+};

spec/operators/isEmpty-spec.js

@@ -0,0 +1,31 @@
+/* globals describe, it, expect, expectObservable, hot */
+var Rx = require('../../dist/cjs/Rx');
+
+describe('Observable.prototype.isEmpty()', function(){
+  it('should return true if source is empty', function () {
+    var source = hot('-----|');
+    var expected =   '-----(x|)';
+
+    expectObservable(source.isEmpty()).toBe(expected, { x: true });
+  });
+
+  it('should return false if source emits element', function () {
+    var source = hot('--a--^--b--|');
+    var expected =        '---(x|)';
+
+    expectObservable(source.isEmpty()).toBe(expected, { x: false });
+  });
+
+  it('should raise error if source raise error', function () {
+    var source = hot('--#');
+    var expected =   '--#';
+
+    expectObservable(source.isEmpty()).toBe(expected);
+  });
+
+  it('should not completes if source never emits', function () {
+    var expected = '-';
+
+    expectObservable(Rx.Observable.never().isEmpty()).toBe(expected);
+  });
+});

src/Observable.ts

@@ -203,6 +203,7 @@ export default class Observable<T>  {
   startWith: <T>(x: T) => Observable<T>;
   debounce: <R>(dueTime: number, scheduler?: Scheduler) => Observable<R>;
 
+  isEmpty: () => Observable<boolean>;
   elementAt: (index: number, defaultValue?: any) => Observable<T>;
   last: (predicate?: (value: T, index:number) => boolean, thisArg?: any, defaultValue?: any) => Observable<T>;
   single: (predicate?: (value: T, index:number) => boolean, thisArg?: any) => Observable<T>;

src/Rx.ts

@@ -221,8 +221,11 @@ import sampleTime from './operators/sampleTime';
 observableProto.sample = sample;
 observableProto.sampleTime = sampleTime;
 
+import isEmpty from './operators/isEmpty';
 import last from './operators/last';
 import single from './operators/single';
+
+observableProto.isEmpty = isEmpty;
 observableProto.last = last;
 observableProto.single = single
 

src/operators/isEmpty.ts

@@ -0,0 +1,35 @@
+import Operator from '../Operator';
+import Observable from '../Observable';
+import Subscriber from '../Subscriber';
+
+export default function isEmpty() {
+  return this.lift(new IsEmptyOperator());
+}
+
+class IsEmptyOperator<T, R> implements Operator<T, R> {
+  call (observer: Subscriber<boolean>): Subscriber<T> {
+    return new IsEmptySubscriber<T>(observer);
+  } 
+}
+
+class IsEmptySubscriber<T> extends Subscriber<T> {
+
+  constructor(destination: Subscriber<boolean>) {
+    super(destination);
+  }
+
+  private notifyComplete(isEmpty: boolean): void {
+    const destination = this.destination;
+
+    destination.next(isEmpty);
+    destination.complete();
+  }
+
+  _next(value: T) {
+    this.notifyComplete(false);
+  }
+
+  _complete() {
+    this.notifyComplete(true);
+  }
+}