feat(operator): Add do operator.

Author: trxcllnt

spec/operators/do-spec.js

@@ -0,0 +1,16 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.do()', function () {
+  it('should do one value', function (done) {
+    var act = false;
+    Observable.value(42).do(function (x) {
+      act = true;
+    })
+    .subscribe(function (x) {
+      expect(x).toBe(42);
+      expect(act).toBe(true);
+    }, null, done);
+  });
+});

src/Observable.ts

@@ -123,6 +123,7 @@ export default class Observable<T> {
   zip: <R>(...observables: (Observable<any> | ((...values: Array<any>) => R)) []) => Observable<R>;
   zipAll: <R>(project?: (...values: Array<any>) => R) => Observable<R>;
 
+  do: <T>(next?: (x: T) => void, error?: (e: any) => void, complete?: () => void) => Observable<T>;
   map: <T, R>(project: (x: T, ix?: number) => R, thisArg?: any) => Observable<R>;
   mapTo: <R>(value: R) => Observable<R>;
   toArray: () => Observable<T[]>;

src/Rx.ts

@@ -75,13 +75,15 @@ observableProto.switchLatest = switchLatest;
 observableProto.switchLatestTo = switchLatestTo;
 observableProto.expand = expand;
 
+import _do from './operators/do';
 import map from './operators/map';
 import mapTo from './operators/mapTo';
 import toArray from './operators/toArray';
 import scan from './operators/scan';
 import reduce from './operators/reduce';
 import startWith from './operators/startWith';
 
+observableProto.do = _do;
 observableProto.map = map;
 observableProto.mapTo = mapTo;
 observableProto.toArray = toArray;

src/operators/do.ts

@@ -0,0 +1,72 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+
+import noop from '../util/noop';
+import tryCatch from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+import bindCallback from '../util/bindCallback';
+
+export default function _do<T>(next?: (x: T) => void, error?: (e: any) => void, complete?: () => void) {
+  return this.lift(new DoOperator(next || noop, error || noop, complete || noop));
+}
+
+export class DoOperator<T, R> extends Operator<T, R> {
+
+  next: (x: T) => void;
+  error: (e: any) => void;
+  complete: () => void;
+
+  constructor(next: (x: T) => void, error: (e: any) => void, complete: () => void) {
+    super();
+    this.next = next;
+    this.error = error;
+    this.complete = complete;
+  }
+
+  call(observer: Observer<T>): Observer<T> {
+    return new DoSubscriber(observer, this.next, this.error, this.complete);
+  }
+}
+
+export class DoSubscriber<T> extends Subscriber<T> {
+
+  __next: (x: T) => void;
+  __error: (e: any) => void;
+  __complete: () => void;
+
+
+  constructor(destination: Observer<T>, next: (x: T) => void, error: (e: any) => void, complete: () => void) {
+    super(destination);
+    this.__next = next;
+    this.__error = error;
+    this.__complete = complete;
+  }
+
+  _next(x) {
+    const result = tryCatch(this.__next)(x);
+    if (result === errorObject) {
+      this.destination.error(errorObject.e);
+    } else {
+      this.destination.next(x);
+    }
+  }
+
+  _error(e) {
+    const result = tryCatch(this.__error)(e);
+    if (result === errorObject) {
+      this.destination.error(errorObject.e);
+    } else {
+      this.destination.error(e);
+    }
+  }
+
+  _complete() {
+    const result = tryCatch(this.__complete)();
+    if (result === errorObject) {
+      this.destination.error(errorObject.e);
+    } else {
+      this.destination.complete();
+    }
+  }
+}