diff --git a/src/operator/reduce.ts b/src/operator/reduce.ts index 7509de24ab..0e6168beec 100644 --- a/src/operator/reduce.ts +++ b/src/operator/reduce.ts @@ -3,38 +3,65 @@ import {Operator} from '../Operator'; import {Subscriber} from '../Subscriber'; /** - * Returns an Observable that applies a specified accumulator function to the first item emitted by a source Observable, - * then feeds the result of that function along with the second item emitted by the source Observable into the same - * function, and so on until all items have been emitted by the source Observable, and emits the final result from - * the final call to your function as its sole item. - * This technique, which is called "reduce" here, is sometimes called "aggregate," "fold," "accumulate," "compress," or - * "inject" in other programming contexts. + * Applies an accumulator function over the source Observable, and returns the + * accumulated result when the source completes, given an optional seed value. + * + * Combines together all values emitted on the source, + * using an accumulator function that knows how to join a new source value into + * the accumulation from the past. * * * - * @param {initialValue} the initial (seed) accumulator value - * @param {accumulator} an accumulator function to be invoked on each item emitted by the source Observable, the - * result of which will be used in the next accumulator call. - * @return {Observable} an Observable that emits a single item that is the result of accumulating the output from the - * items emitted by the source Observable. + * Like + * [Array.prototype.reduce()](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/reduce), + * `reduce` applies an `accumulator` function against an accumulation and each + * value of the source Observable (from the past) to reduce it to a single + * value, emitted on the output Observable. Note that `reduce` will only emit + * one value, only when the source Observable completes. It is equivalent to + * applying operator {@link scan} followed by operator {@link last}. + * + * Returns an Observable that applies a specified `accumulator` function to each + * item emitted by the source Observable. If a `seed` value is specified, then + * that value will be used as the initial value for the accumulator. If no seed + * value is specified, the first item of the source is used as the seed. + * + * @example Count the number of click events that happened in 5 seconds + * var clicksInFiveSeconds = Rx.Observable.fromEvent(document, 'click') + * .takeUntil(Rx.Observable.interval(5000)); + * var ones = clicksInFiveSeconds.mapTo(1); + * var seed = 0; + * var count = ones.reduce((acc, one) => acc + one, seed); + * count.subscribe(x => console.log(x)); + * + * @see {@link count} + * @see {@link expand} + * @see {@link mergeScan} + * @see {@link scan} + * + * @param {function(acc: R, value: T): R} accumulator The accumulator function + * called on each source value. + * @param {R} [seed] The initial accumulation value. + * @return {Observable} An observable of the accumulated values. + * @return {Observable} An Observable that emits a single value that is the + * result of accumulating the values emitted by the source Observable. * @method reduce * @owner Observable */ -export function reduce(project: (acc: R, value: T) => R, seed?: R): Observable { - return this.lift(new ReduceOperator(project, seed)); +export function reduce(accumulator: (acc: R, value: T) => R, seed?: R): Observable { + return this.lift(new ReduceOperator(accumulator, seed)); } export interface ReduceSignature { - (project: (acc: R, value: T) => R, seed?: R): Observable; + (accumulator: (acc: R, value: T) => R, seed?: R): Observable; } export class ReduceOperator implements Operator { - constructor(private project: (acc: R, value: T) => R, private seed?: R) { + constructor(private accumulator: (acc: R, value: T) => R, private seed?: R) { } call(subscriber: Subscriber, source: any): any { - return source._subscribe(new ReduceSubscriber(subscriber, this.project, this.seed)); + return source._subscribe(new ReduceSubscriber(subscriber, this.accumulator, this.seed)); } } @@ -48,12 +75,13 @@ export class ReduceSubscriber extends Subscriber { acc: T | R; hasSeed: boolean; hasValue: boolean = false; - project: (acc: R, value: T) => R; - constructor(destination: Subscriber, project: (acc: R, value: T) => R, seed?: R) { + constructor(destination: Subscriber, + private accumulator: (acc: R, value: T) => R, + seed?: R) { super(destination); this.acc = seed; - this.project = project; + this.accumulator = accumulator; this.hasSeed = typeof seed !== 'undefined'; } @@ -69,7 +97,7 @@ export class ReduceSubscriber extends Subscriber { private _tryReduce(value: T) { let result: any; try { - result = this.project(this.acc, value); + result = this.accumulator(this.acc, value); } catch (err) { this.destination.error(err); return; diff --git a/src/operator/scan.ts b/src/operator/scan.ts index 61a7e1e80b..a85ea03fa8 100644 --- a/src/operator/scan.ts +++ b/src/operator/scan.ts @@ -3,7 +3,7 @@ import {Observable} from '../Observable'; import {Subscriber} from '../Subscriber'; /** - * Applies an accumulation function over the source Observable, and returns each + * Applies an accumulator function over the source Observable, and returns each * intermediate result, with an optional seed value. * * It's like {@link reduce}, but emits the current