feat(operator): Add count operator.

Author: trxcllnt

spec/operators/count-spec.js

@@ -0,0 +1,13 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('count', function () {
+  it('should count the values of an observable', function (done) {
+    Observable.fromArray([1, 2, 3])
+      .count()
+      .subscribe(function (total) {
+          expect(total).toEqual(3);
+        }, null, done);
+  });
+});

spec/operators/takeUntil-spec.js

@@ -8,8 +8,8 @@ describe('Observable.prototype.takeUntil()', function () {
     var i = 0;
     var nextSpy = jasmine.createSpy('nextSpy');
 
-    Observable.timer(0, 16)
-      .takeUntil(Observable.timer(81))
+    Observable.timer(0, 100)
+      .takeUntil(Observable.timer(450))
       .subscribe(nextSpy, null, function () {
         expect(nextSpy.calls.count()).toBe(5);
         expected.forEach(function (v) {

src/Observable.ts

@@ -129,6 +129,7 @@ export default class Observable<T> {
   map: <T, R>(project: (x: T, ix?: number) => R, thisArg?: any) => Observable<R>;
   mapTo: <R>(value: R) => Observable<R>;
   toArray: () => Observable<T[]>;
+  count: () => Observable<number>;
   scan: <R>(project: (acc: R, x: T) => R, acc?: R) => Observable<R>;
   reduce: <R>(project: (acc: R, x: T) => R, acc?: R) => Observable<R>;
   startWith: <T>(x: T) => Observable<T>;

src/Rx.ts

@@ -79,6 +79,7 @@ import _do from './operators/do';
 import map from './operators/map';
 import mapTo from './operators/mapTo';
 import toArray from './operators/toArray';
+import count from './operators/count';
 import scan from './operators/scan';
 import reduce from './operators/reduce';
 import startWith from './operators/startWith';
@@ -87,6 +88,7 @@ observableProto.do = _do;
 observableProto.map = map;
 observableProto.mapTo = mapTo;
 observableProto.toArray = toArray;
+observableProto.count = count;
 observableProto.scan = scan;
 observableProto.reduce = reduce;
 observableProto.startWith = startWith;

src/observables/RangeObservable.ts

@@ -3,15 +3,15 @@ import Observable from '../Observable';
 
 export default class RangeObservable<T> extends Observable<T> {
 
-  static create(start: number = 0, count: number = 0, scheduler?: Scheduler) {
-    return new RangeObservable(start, count, scheduler);
+  static create(start: number = 0, end: number = 0, scheduler?: Scheduler) {
+    return new RangeObservable(start, end, scheduler);
   }
 
   static dispatch(state) {
 
-    const { start, index, count, subscriber } = state;
+    const { start, index, end, subscriber } = state;
 
-    if (index >= count) {
+    if (index >= end) {
       subscriber.complete();
       return;
     }
@@ -28,24 +28,31 @@ export default class RangeObservable<T> extends Observable<T> {
     (<any> this).schedule(state);
   }
 
-  constructor(private start: number, private count: number, private scheduler?: Scheduler) {
+  private start: number;
+  private end: number;
+  private scheduler: Scheduler;
+
+  constructor(start: number, end: number, scheduler?: Scheduler) {
     super();
+    this.start = start;
+    this.end = end;
+    this.scheduler = scheduler;
   }
 
   _subscribe(subscriber) {
 
     let index = 0;
     let start = this.start;
-    const count = this.count;
+    const end = this.end;
     const scheduler = this.scheduler;
 
     if (scheduler) {
       subscriber.add(scheduler.schedule(0, {
-        index, count, start, subscriber
+        index, end, start, subscriber
       }, RangeObservable.dispatch));
     } else {
       do {
-        if (index++ >= count) {
+        if (index++ >= end) {
           subscriber.complete();
           break;
         }

src/operators/count.ts

@@ -0,0 +1,31 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+
+export default function count() {
+  return this.lift(new CountOperator());
+}
+
+export class CountOperator<T, R> extends Operator<T, R> {
+  call(observer: Observer<number>): Observer<T> {
+    return new CountSubscriber<T>(observer);
+  }
+}
+
+export class CountSubscriber<T> extends Subscriber<T> {
+
+  count: number = 0;
+
+  constructor(destination: Observer<number>) {
+    super(destination);
+  }
+
+  _next(x) {
+    this.count += 1;
+  }
+
+  _complete() {
+    this.destination.next(this.count);
+    this.destination.complete();
+  }
+}