feat(operator): add ignoreElements operator

Author: jinroh

Diff for: perf/micro/current-thread-scheduler/operators/ignore-elements.js

@@ -0,0 +1,20 @@
+var RxOld = require("rx");
+var RxNew = require("../../../../index");
+
+module.exports = function (suite) {
+
+    var oldIgnoreElementsWithCurrentThreadScheduler = RxOld.Observable.range(0, 50, RxOld.Scheduler.currentThread).ignoreElements();
+    var newIgnoreElementsWithCurrentThreadScheduler = RxNew.Observable.range(0, 50, RxNew.Scheduler.immediate).ignoreElements();
+
+    return suite
+        .add('old ignoreElements with current thread scheduler', function () {
+            oldIgnoreElementsWithCurrentThreadScheduler.subscribe(_next, _error, _complete);
+        })
+        .add('new ignoreElements with current thread scheduler', function () {
+            newIgnoreElementsWithCurrentThreadScheduler.subscribe(_next, _error, _complete);
+        });
+
+    function _next(x) { }
+    function _error(e){ }
+    function _complete(){ }
+};

Diff for: perf/micro/immediate-scheduler/operators/ignore-elements.js

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

Diff for: spec/operators/ignoreElements-spec.js

@@ -0,0 +1,40 @@
+/* globals describe, it, expect, hot, cold, expectObservable */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.ignoreElements', function() {
+  it('should ignore all the elements of the source', function() {
+    var source = hot('--a--b--c--d--|');
+    var expected =   '--------------|';
+
+    expectObservable(source.ignoreElements()).toBe(expected);
+  });
+
+  it('should propagate errors from the source', function() {
+    var source = hot('--a--#');
+    var expected =   '-----#';
+
+    expectObservable(source.ignoreElements()).toBe(expected);
+  });
+
+  it('should support Observable.empty', function() {
+    var source = Observable.empty();
+    var expected = "|";
+
+    expectObservable(source.ignoreElements()).toBe(expected);
+  });
+
+  it('should support Observable.never', function() {
+    var source = Observable.never();
+    var expected = "-";
+
+    expectObservable(source.ignoreElements()).toBe(expected);
+  });
+
+  it('should support Observable.throw', function() {
+    var source = Observable.throw(new Error('oops'));
+    var expected = '#';
+
+    expectObservable(source.ignoreElements()).toBe(expected, undefined, new Error('oops'));
+  });
+});

Diff for: src/Observable.ts

@@ -211,6 +211,7 @@ export default class Observable<T>  {
   filter: (predicate: (x: T) => boolean, ix?: number, thisArg?: any) => Observable<T>;
   distinctUntilChanged: (compare?: (x: T, y: T) => boolean, thisArg?: any) => Observable<T>;
   distinctUntilKeyChanged: (key: string, compare?: (x: any, y: any) => boolean, thisArg?: any) => Observable<T>;
+  ignoreElements: () => Observable<T>;
   skip: (count: number) => Observable<T>;
   skipUntil: (notifier: Observable<any>) => Observable<T>;
   take: (count: number) => Observable<T>;

Diff for: src/Rx.ts

@@ -109,6 +109,7 @@ import elementAt from './operators/elementAt';
 import filter from './operators/filter';
 import distinctUntilChanged from './operators/distinctUntilChanged';
 import distinctUntilKeyChanged from './operators/distinctUntilKeyChanged';
+import ignoreElements from './operators/ignoreElements';
 
 observableProto.take = take;
 observableProto.skip = skip;
@@ -118,6 +119,7 @@ observableProto.elementAt = elementAt;
 observableProto.filter = filter;
 observableProto.distinctUntilChanged = distinctUntilChanged;
 observableProto.distinctUntilKeyChanged = distinctUntilKeyChanged;
+observableProto.ignoreElements = ignoreElements;
 
 import combineLatestProto from './operators/combineLatest';
 import combineLatest from './operators/combineLatest-static';

Diff for: src/operators/ignoreElements.ts

@@ -0,0 +1,16 @@
+import Operator from '../Operator';
+import Subscriber from '../Subscriber';
+
+export default function ignoreElements() {
+  return this.lift(new IgnoreElementsOperator());
+};
+
+class IgnoreElementsOperator<T, R> implements Operator<T, R> {
+  call(subscriber: Subscriber<R>): Subscriber<T> {
+    return new IgnoreElementsSubscriber(subscriber);
+  }
+}
+
+class IgnoreElementsSubscriber<T> extends Subscriber<T> {
+  _next() {}
+}