feat(fp): add fp operators

generates the new operators using the existing prototype operators, and a script to generate the resulting fp types.

Author: david-driscoll

package.json

@@ -56,9 +56,9 @@
     "commitmsg": "validate-commit-msg",
     "info": "npm-scripts-info",
     "build_all": "npm-run-all build_cjs build_global generate_packages",
-    "build_cjs": "npm-run-all clean_dist_cjs copy_src_cjs compile_dist_cjs",
-    "build_es6": "npm-run-all clean_dist_es6 copy_src_es6 compile_module_es6",
-    "build_es6_for_docs": "npm-run-all clean_dist_es6 copy_src_es6 compile_dist_es6_for_docs",
+    "build_cjs": "npm-run-all generate_fp clean_dist_cjs copy_src_cjs compile_dist_cjs",
+    "build_es6": "npm-run-all generate_fp clean_dist_es6 copy_src_es6 compile_module_es6",
+    "build_es6_for_docs": "npm-run-all generate_fp clean_dist_es6 copy_src_es6 compile_dist_es6_for_docs",
     "build_closure_core": "node ./tools/make-closure-core.js",
     "build_global": "npm-run-all clean_dist_global build_es6 && mkdirp ./dist/global && node ./tools/make-umd-bundle.js && npm-run-all build_closure_core clean_dist_es6",
     "build_perf": "webdriver-manager update && npm-run-all build_cjs build_global perf",
@@ -75,13 +75,14 @@
     "copy_src_cjs": "mkdirp ./dist/cjs/src && shx cp -r ./src/* ./dist/cjs/src",
     "copy_src_es6": "mkdirp ./dist/es6/src && shx cp -r ./src/* ./dist/es6/src",
     "commit": "git-cz",
-    "compile_dist_cjs": "tsc ./dist/cjs/src/Rx.ts ./dist/cjs/src/add/observable/of.ts -m commonjs --lib es5,es2015.iterable,es2015.collection,es2015.promise,dom --sourceMap --outDir ./dist/cjs --target ES5 -d --diagnostics --pretty --noImplicitAny --noImplicitReturns --noImplicitThis --suppressImplicitAnyIndexErrors --moduleResolution node",
-    "compile_module_es6": "tsc                                          ./dist/es6/src/Rx.ts ./dist/es6/src/add/observable/of.ts -m es2015   --sourceMap --outDir ./dist/es6 --target ES5 -d --diagnostics --pretty --noImplicitAny --noImplicitReturns --noImplicitThis --suppressImplicitAnyIndexErrors --moduleResolution node --noEmitHelpers --lib es5,es2015.iterable,es2015.collection,es2015.promise,dom ",
-    "compile_dist_es6_for_docs": "tsc                                 ./dist/es6/src/Rx.ts ./dist/es6/src/add/observable/of.ts ./dist/es6/src/MiscJSDoc.ts -m es2015   --sourceMap --outDir ./dist/es6 --target ES6 -d --diagnostics --pretty --noImplicitAny --noImplicitReturns --noImplicitThis --suppressImplicitAnyIndexErrors --moduleResolution node",
-    "cover": "shx rm -rf dist/cjs && tsc src/Rx.ts src/add/observable/of.ts -m commonjs --lib es5,es2015.iterable,es2015.collection,es2015.promise,dom --outDir dist/cjs --sourceMap --target ES5 -d && nyc --reporter=lcov --reporter=html --exclude=spec/support/**/* --exclude=spec-js/**/* --exclude=node_modules mocha --opts spec/support/default.opts spec-js",
+    "compile_dist_cjs": "tsc ./dist/cjs/src/Rx.ts ./dist/cjs/src/fp.ts ./dist/cjs/src/add/observable/of.ts -m commonjs --lib es5,es2015.iterable,es2015.collection,es2015.promise,dom --sourceMap --outDir ./dist/cjs --target ES5 -d --diagnostics --pretty --noImplicitAny --noImplicitReturns --noImplicitThis --suppressImplicitAnyIndexErrors --moduleResolution node",
+    "compile_module_es6": "tsc                                          ./dist/es6/src/Rx.ts ./dist/es6/src/fp.ts ./dist/es6/src/add/observable/of.ts -m es2015   --sourceMap --outDir ./dist/es6 --target ES5 -d --diagnostics --pretty --noImplicitAny --noImplicitReturns --noImplicitThis --suppressImplicitAnyIndexErrors --moduleResolution node --noEmitHelpers --lib es5,es2015.iterable,es2015.collection,es2015.promise,dom ",
+    "compile_dist_es6_for_docs": "tsc                                 ./dist/es6/src/Rx.ts ./dist/es6/src/fp.ts ./dist/es6/src/add/observable/of.ts ./dist/es6/src/MiscJSDoc.ts -m es2015   --sourceMap --outDir ./dist/es6 --target ES6 -d --diagnostics --pretty --noImplicitAny --noImplicitReturns --noImplicitThis --suppressImplicitAnyIndexErrors --moduleResolution node",
+    "cover": "shx rm -rf dist/cjs && tsc src/Rx.ts src/fp.ts src/add/observable/of.ts -m commonjs --lib es5,es2015.iterable,es2015.collection,es2015.promise,dom --outDir dist/cjs --sourceMap --target ES5 -d && nyc --reporter=lcov --reporter=html --exclude=spec/support/**/* --exclude=spec-js/**/* --exclude=node_modules mocha --opts spec/support/default.opts spec-js",
     "decision_tree_widget": "cd doc/decision-tree-widget && npm run build && cd ../..",
     "doctoc": "doctoc CONTRIBUTING.md",
     "generate_packages": "node .make-packages.js",
+    "generate_fp": "node tools/make-fp.js",
     "lint_perf": "eslint perf/",
     "lint_spec": "tslint -c tslint.json \"spec/**/*.ts\"",
     "lint_src": "tslint -c tslint.json \"src/**/*.ts\"",

spec/Observable-spec.ts

@@ -656,7 +656,7 @@ describe('Observable.create', () => {
 
   describe('pipe', () => {
     it('should exist', () => {
-      expect(Observable.prototype.pipe).to.exist;
+      expect(Observable.prototype.let).to.exist;
     });
 
     it('should pipe through multiple functions', () => {
@@ -666,7 +666,7 @@ describe('Observable.create', () => {
       const ob4 = Observable.of(4);
       const results = [];
 
-      const piped = ob1.pipe(
+      const piped = ob1.let(
         s => {
           expect(s).to.equal(ob1);
           return ob2;

spec/operators/groupBy-spec.ts

@@ -1,6 +1,6 @@
 import {expect} from 'chai';
 import * as Rx from '../../dist/cjs/Rx';
-import {GroupedObservable} from '../../dist/cjs/operator/groupBy';
+import {GroupedObservable} from '../../dist/cjs/observable/GroupedObservable';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
 
 declare const { asDiagram };

spec/pipe/map-spec.ts

@@ -8,14 +8,14 @@ declare const cold: typeof marbleTestingSignature.cold;
 declare const expectObservable: typeof marbleTestingSignature.expectObservable;
 declare const expectSubscriptions: typeof marbleTestingSignature.expectSubscriptions;
 
-describe('map-pipe', () => {
-  it('should pipe through the map operator', () => {
+describe('map-let', () => {
+  it('should let through the map operator', () => {
     const s1 = cold( '---a--b--c--|');
     const project = x => x + x;
-    expectObservable(Rx.Pipe.map(project)(s1)).toBe(s1.map(project));
+    expectObservable(Rx.fp.map(project)(s1)).toBe(s1.map(project));
   });
 
-  it('should work with Observable.prototype.pipe', () => {
+  it('should work with Observable.prototype.let', () => {
     const values = {
       a: 1,
       b: 2,
@@ -27,9 +27,9 @@ describe('map-pipe', () => {
 
     const s1 = cold( '---a---b---c---|', values);
     const expected = '---x---y---z---|';
-    const { map } = Rx.Pipe;
+    const { map } = Rx.fp;
 
-    const t1 = s1.pipe(
+    const t1 = s1.let(
       map((x: number) => x + 1),
       map((x: number) => x / 2)
     );

src/Observable.ts

@@ -7,7 +7,6 @@ import { toSubscriber } from './util/toSubscriber';
 import { IfObservable } from './observable/IfObservable';
 import { ErrorObservable } from './observable/ErrorObservable';
 import { observable as Symbol_observable } from './symbol/observable';
-import { compose } from './util/compose';
 import { letProto } from './operator/let';
 
 export interface Subscribable<T> {
@@ -171,11 +170,6 @@ export class Observable<T> implements Subscribable<T> {
     });
   }
 
-  pipe(...fns: ((x: Observable<T>) => Observable<T>)[]): Observable<T> {
-    const composed = compose.apply(this, fns);
-    return composed(this);
-  }
-
   protected _subscribe(subscriber: Subscriber<any>): TeardownLogic {
     return this.source.subscribe(subscriber);
   }

src/Rx.ts

@@ -176,15 +176,8 @@ import { observable } from './symbol/observable';
 /* tslint:enable:no-unused-variable */
 export { compose } from './util/compose';
 
-// pull in pipe definitions
-import { map as mapPipe } from './pipe/map';
-
-/**
- * @typedef {Object} Rx.Pipe
- */
-let Pipe = {
-  map: mapPipe
-};
+// pull in fp definitions
+import * as fp from './fp';
 
 /**
  * @typedef {Object} Rx.Scheduler
@@ -228,6 +221,6 @@ let Symbol = {
 export {
     Scheduler,
     Symbol,
-    Pipe,
+    fp,
     Observable
 };

src/fp.ts

@@ -0,0 +1,100 @@
+export { audit } from './fp/audit';
+export { auditTime } from './fp/auditTime';
+export { buffer } from './fp/buffer';
+export { bufferCount } from './fp/bufferCount';
+export { bufferTime } from './fp/bufferTime';
+export { bufferToggle } from './fp/bufferToggle';
+export { bufferWhen } from './fp/bufferWhen';
+export { _catch } from './fp/catch';
+export { combineAll } from './fp/combineAll';
+export { combineLatest } from './fp/combineLatest';
+export { concat } from './fp/concat';
+export { concatAll } from './fp/concatAll';
+export { concatMap } from './fp/concatMap';
+export { concatMapTo } from './fp/concatMapTo';
+export { count } from './fp/count';
+export { debounce } from './fp/debounce';
+export { debounceTime } from './fp/debounceTime';
+export { defaultIfEmpty } from './fp/defaultIfEmpty';
+export { delay } from './fp/delay';
+export { delayWhen } from './fp/delayWhen';
+export { dematerialize } from './fp/dematerialize';
+export { distinct } from './fp/distinct';
+export { distinctUntilChanged } from './fp/distinctUntilChanged';
+export { distinctUntilKeyChanged } from './fp/distinctUntilKeyChanged';
+export { _do } from './fp/do';
+export { elementAt } from './fp/elementAt';
+export { every } from './fp/every';
+export { exhaust } from './fp/exhaust';
+export { exhaustMap } from './fp/exhaustMap';
+export { expand } from './fp/expand';
+export { filter } from './fp/filter';
+export { _finally } from './fp/finally';
+export { find } from './fp/find';
+export { findIndex } from './fp/findIndex';
+export { first } from './fp/first';
+export { groupBy } from './fp/groupBy';
+export { ignoreElements } from './fp/ignoreElements';
+export { isEmpty } from './fp/isEmpty';
+export { last } from './fp/last';
+export { letProto } from './fp/let';
+export { map } from './fp/map';
+export { mapTo } from './fp/mapTo';
+export { materialize } from './fp/materialize';
+export { max } from './fp/max';
+export { merge } from './fp/merge';
+export { mergeAll } from './fp/mergeAll';
+export { mergeMap } from './fp/mergeMap';
+export { mergeMapTo } from './fp/mergeMapTo';
+export { mergeScan } from './fp/mergeScan';
+export { min } from './fp/min';
+export { multicast } from './fp/multicast';
+export { observeOn } from './fp/observeOn';
+export { onErrorResumeNext } from './fp/onErrorResumeNext';
+export { pairwise } from './fp/pairwise';
+export { partition } from './fp/partition';
+export { pluck } from './fp/pluck';
+export { publish } from './fp/publish';
+export { publishBehavior } from './fp/publishBehavior';
+export { publishLast } from './fp/publishLast';
+export { publishReplay } from './fp/publishReplay';
+export { race } from './fp/race';
+export { reduce } from './fp/reduce';
+export { repeat } from './fp/repeat';
+export { repeatWhen } from './fp/repeatWhen';
+export { retry } from './fp/retry';
+export { retryWhen } from './fp/retryWhen';
+export { sample } from './fp/sample';
+export { sampleTime } from './fp/sampleTime';
+export { scan } from './fp/scan';
+export { sequenceEqual } from './fp/sequenceEqual';
+export { share } from './fp/share';
+export { single } from './fp/single';
+export { skip } from './fp/skip';
+export { skipUntil } from './fp/skipUntil';
+export { skipWhile } from './fp/skipWhile';
+export { startWith } from './fp/startWith';
+export { subscribeOn } from './fp/subscribeOn';
+export { _switch } from './fp/switch';
+export { switchMap } from './fp/switchMap';
+export { switchMapTo } from './fp/switchMapTo';
+export { take } from './fp/take';
+export { takeLast } from './fp/takeLast';
+export { takeUntil } from './fp/takeUntil';
+export { takeWhile } from './fp/takeWhile';
+export { throttle } from './fp/throttle';
+export { throttleTime } from './fp/throttleTime';
+export { timeInterval } from './fp/timeInterval';
+export { timeout } from './fp/timeout';
+export { timeoutWith } from './fp/timeoutWith';
+export { timestamp } from './fp/timestamp';
+export { toArray } from './fp/toArray';
+export { toPromise } from './fp/toPromise';
+export { window } from './fp/window';
+export { windowCount } from './fp/windowCount';
+export { windowTime } from './fp/windowTime';
+export { windowToggle } from './fp/windowToggle';
+export { windowWhen } from './fp/windowWhen';
+export { withLatestFrom } from './fp/withLatestFrom';
+export { zipProto } from './fp/zip';
+export { zipAll } from './fp/zipAll';

src/fp/audit.ts

@@ -0,0 +1,55 @@
+// Generated code ahead... there be dragons!
+// tslint:disable
+import { audit as auditBase } from '../operator/audit';
+import { Subscriber } from '../Subscriber';
+import { Observable, SubscribableOrPromise } from '../Observable';
+import { Subscription, TeardownLogic } from '../Subscription';
+
+import { tryCatch } from '../util/tryCatch';
+import { errorObject } from '../util/errorObject';
+import { OuterSubscriber } from '../OuterSubscriber';
+import { subscribeToResult } from '../util/subscribeToResult';
+
+/**
+ * Ignores source values for a duration determined by another Observable, then
+ * emits the most recent value from the source Observable, then repeats this
+ * process.
+ *
+ * <span class="informal">It's like {@link auditTime}, but the silencing
+ * duration is determined by a second Observable.</span>
+ *
+ * <img src="./img/audit.png" width="100%">
+ *
+ * `audit` is similar to `throttle`, but emits the last value from the silenced
+ * time window, instead of the first value. `audit` emits the most recent value
+ * from the source Observable on the output Observable as soon as its internal
+ * timer becomes disabled, and ignores source values while the timer is enabled.
+ * Initially, the timer is disabled. As soon as the first source value arrives,
+ * the timer is enabled by calling the `durationSelector` function with the
+ * source value, which returns the "duration" Observable. When the duration
+ * Observable emits a value or completes, the timer is disabled, then the most
+ * recent source value is emitted on the output Observable, and this process
+ * repeats for the next source value.
+ *
+ * @example <caption>Emit clicks at a rate of at most one click per second</caption>
+ * var clicks = Rx.Observable.fromEvent(document, 'click');
+ * var result = clicks.audit(ev => Rx.Observable.interval(1000));
+ * result.subscribe(x => console.log(x));
+ *
+ * @see {@link auditTime}
+ * @see {@link debounce}
+ * @see {@link delayWhen}
+ * @see {@link sample}
+ * @see {@link throttle}
+ *
+ * @param {function(value: T): SubscribableOrPromise} durationSelector A function
+ * that receives a value from the source Observable, for computing the silencing
+ * duration, returned as an Observable or a Promise.
+ * @return {Observable<T>} An Observable that performs rate-limiting of
+ * emissions from the source Observable.
+ * @method audit
+ * @owner Observable
+ */
+export function audit<T>( durationSelector: (value: T) => SubscribableOrPromise<any>): (source: Observable<T>) => Observable<T> {
+  return (source: Observable<T>) => auditBase.call(source, durationSelector);
+}

src/fp/auditTime.ts

@@ -0,0 +1,54 @@
+// Generated code ahead... there be dragons!
+// tslint:disable
+import { auditTime as auditTimeBase } from '../operator/auditTime';
+import { async } from '../scheduler/async';
+import { IScheduler } from '../Scheduler';
+import { Subscriber } from '../Subscriber';
+import { Observable } from '../Observable';
+import { Subscription, TeardownLogic } from '../Subscription';
+
+/**
+ * Ignores source values for `duration` milliseconds, then emits the most recent
+ * value from the source Observable, then repeats this process.
+ *
+ * <span class="informal">When it sees a source values, it ignores that plus
+ * the next ones for `duration` milliseconds, and then it emits the most recent
+ * value from the source.</span>
+ *
+ * <img src="./img/auditTime.png" width="100%">
+ *
+ * `auditTime` is similar to `throttleTime`, but emits the last value from the
+ * silenced time window, instead of the first value. `auditTime` emits the most
+ * recent value from the source Observable on the output Observable as soon as
+ * its internal timer becomes disabled, and ignores source values while the
+ * timer is enabled. Initially, the timer is disabled. As soon as the first
+ * source value arrives, the timer is enabled. After `duration` milliseconds (or
+ * the time unit determined internally by the optional `scheduler`) has passed,
+ * the timer is disabled, then the most recent source value is emitted on the
+ * output Observable, and this process repeats for the next source value.
+ * Optionally takes a {@link IScheduler} for managing timers.
+ *
+ * @example <caption>Emit clicks at a rate of at most one click per second</caption>
+ * var clicks = Rx.Observable.fromEvent(document, 'click');
+ * var result = clicks.auditTime(1000);
+ * result.subscribe(x => console.log(x));
+ *
+ * @see {@link audit}
+ * @see {@link debounceTime}
+ * @see {@link delay}
+ * @see {@link sampleTime}
+ * @see {@link throttleTime}
+ *
+ * @param {number} duration Time to wait before emitting the most recent source
+ * value, measured in milliseconds or the time unit determined internally
+ * by the optional `scheduler`.
+ * @param {Scheduler} [scheduler=async] The {@link IScheduler} to use for
+ * managing the timers that handle the rate-limiting behavior.
+ * @return {Observable<T>} An Observable that performs rate-limiting of
+ * emissions from the source Observable.
+ * @method auditTime
+ * @owner Observable
+ */
+export function auditTime<T>( duration: number, scheduler: IScheduler = async): (source: Observable<T>) => Observable<T> {
+  return (source: Observable<T>) => auditTimeBase.call(source, duration, scheduler);
+}

src/fp/buffer.ts

@@ -0,0 +1,45 @@
+// Generated code ahead... there be dragons!
+// tslint:disable
+import { buffer as bufferBase } from '../operator/buffer';
+import { Subscriber } from '../Subscriber';
+import { Observable } from '../Observable';
+
+import { OuterSubscriber } from '../OuterSubscriber';
+import { InnerSubscriber } from '../InnerSubscriber';
+import { subscribeToResult } from '../util/subscribeToResult';
+
+/**
+ * Buffers the source Observable values until `closingNotifier` emits.
+ *
+ * <span class="informal">Collects values from the past as an array, and emits
+ * that array only when another Observable emits.</span>
+ *
+ * <img src="./img/buffer.png" width="100%">
+ *
+ * Buffers the incoming Observable values until the given `closingNotifier`
+ * Observable emits a value, at which point it emits the buffer on the output
+ * Observable and starts a new buffer internally, awaiting the next time
+ * `closingNotifier` emits.
+ *
+ * @example <caption>On every click, emit array of most recent interval events</caption>
+ * var clicks = Rx.Observable.fromEvent(document, 'click');
+ * var interval = Rx.Observable.interval(1000);
+ * var buffered = interval.buffer(clicks);
+ * buffered.subscribe(x => console.log(x));
+ *
+ * @see {@link bufferCount}
+ * @see {@link bufferTime}
+ * @see {@link bufferToggle}
+ * @see {@link bufferWhen}
+ * @see {@link window}
+ *
+ * @param {Observable<any>} closingNotifier An Observable that signals the
+ * buffer to be emitted on the output Observable.
+ * @return {Observable<T[]>} An Observable of buffers, which are arrays of
+ * values.
+ * @method buffer
+ * @owner Observable
+ */
+export function buffer<T>( closingNotifier: Observable<any>): (source: Observable<T>) => Observable<T[]> {
+  return (source: Observable<T>) => bufferBase.call(source, closingNotifier);
+}