diff --git a/src/operator/pairwise.ts b/src/operator/pairwise.ts index cc97da46f7..e17d8db403 100644 --- a/src/operator/pairwise.ts +++ b/src/operator/pairwise.ts @@ -1,6 +1,5 @@ -import { Operator } from '../Operator'; import { Observable } from '../Observable'; -import { Subscriber } from '../Subscriber'; +import { pairwise as higherOrder } from '../operators/pairwise'; /** * Groups pairs of consecutive emissions together and emits them as an array of @@ -38,35 +37,5 @@ import { Subscriber } from '../Subscriber'; * @owner Observable */ export function pairwise(this: Observable): Observable<[T, T]> { - return this.lift(new PairwiseOperator()); -} - -class PairwiseOperator implements Operator { - call(subscriber: Subscriber<[T, T]>, source: any): any { - return source.subscribe(new PairwiseSubscriber(subscriber)); - } -} - -/** - * We need this JSDoc comment for affecting ESDoc. - * @ignore - * @extends {Ignored} - */ -class PairwiseSubscriber extends Subscriber { - private prev: T; - private hasPrev: boolean = false; - - constructor(destination: Subscriber<[T, T]>) { - super(destination); - } - - _next(value: T): void { - if (this.hasPrev) { - this.destination.next([this.prev, value]); - } else { - this.hasPrev = true; - } - - this.prev = value; - } + return higherOrder()(this); } diff --git a/src/operators/index.ts b/src/operators/index.ts index 71cd379730..01a74a9f01 100644 --- a/src/operators/index.ts +++ b/src/operators/index.ts @@ -46,6 +46,7 @@ export { min } from './min'; export { multicast } from './multicast'; export { observeOn } from './observeOn'; export { onErrorResumeNext } from './onErrorResumeNext'; +export { pairwise } from './pairwise'; export { publish } from './publish'; export { race } from './race'; export { reduce } from './reduce'; diff --git a/src/operators/pairwise.ts b/src/operators/pairwise.ts new file mode 100644 index 0000000000..8a9657398c --- /dev/null +++ b/src/operators/pairwise.ts @@ -0,0 +1,73 @@ +import { Operator } from '../Operator'; +import { Observable } from '../Observable'; +import { Subscriber } from '../Subscriber'; +import { OperatorFunction } from '../interfaces'; + +/** + * Groups pairs of consecutive emissions together and emits them as an array of + * two values. + * + * Puts the current value and previous value together as + * an array, and emits that. + * + * + * + * The Nth emission from the source Observable will cause the output Observable + * to emit an array [(N-1)th, Nth] of the previous and the current value, as a + * pair. For this reason, `pairwise` emits on the second and subsequent + * emissions from the source Observable, but not on the first emission, because + * there is no previous value in that case. + * + * @example On every click (starting from the second), emit the relative distance to the previous click + * var clicks = Rx.Observable.fromEvent(document, 'click'); + * var pairs = clicks.pairwise(); + * var distance = pairs.map(pair => { + * var x0 = pair[0].clientX; + * var y0 = pair[0].clientY; + * var x1 = pair[1].clientX; + * var y1 = pair[1].clientY; + * return Math.sqrt(Math.pow(x0 - x1, 2) + Math.pow(y0 - y1, 2)); + * }); + * distance.subscribe(x => console.log(x)); + * + * @see {@link buffer} + * @see {@link bufferCount} + * + * @return {Observable>} An Observable of pairs (as arrays) of + * consecutive values from the source Observable. + * @method pairwise + * @owner Observable + */ +export function pairwise(): OperatorFunction { + return (source: Observable) => source.lift(new PairwiseOperator()); +} + +class PairwiseOperator implements Operator { + call(subscriber: Subscriber<[T, T]>, source: any): any { + return source.subscribe(new PairwiseSubscriber(subscriber)); + } +} + +/** + * We need this JSDoc comment for affecting ESDoc. + * @ignore + * @extends {Ignored} + */ +class PairwiseSubscriber extends Subscriber { + private prev: T; + private hasPrev: boolean = false; + + constructor(destination: Subscriber<[T, T]>) { + super(destination); + } + + _next(value: T): void { + if (this.hasPrev) { + this.destination.next([this.prev, value]); + } else { + this.hasPrev = true; + } + + this.prev = value; + } +}