feat(mergeScan): add new mergeScan operator.

spec/operators/mergeScan-spec.js

@@ -0,0 +1,106 @@
+/* globals describe, it, expect, expectObservable, hot */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.mergeScan()', function () {
+ it('should mergeScan things', function () {
+ var e1 = hot('--a--^--b--c--d--e--f--g--|');
+ var expected = '---u--v--w--x--y--z--|';
+
+ var values = {
+ u: ['b'],
+ v: ['b', 'c'],
+ w: ['b', 'c', 'd'],
+ x: ['b', 'c', 'd', 'e'],
+ y: ['b', 'c', 'd', 'e', 'f'],
+ z: ['b', 'c', 'd', 'e', 'f', 'g']
+ };
+
+ var source = e1.mergeScan(function (acc, x) { return Observable.of(acc.concat(x)); }, []);
+
+ expectObservable(source).toBe(expected, values);
+ });
+
+ it('should handle errors', function () {
+ var e1 = hot('--a--^--b--c--d--#');
+ var expected = '---u--v--w--#';
+
+ var values = {
+ u: ['b'],
+ v: ['b', 'c'],
+ w: ['b', 'c', 'd']
+ };
+
+ var source = e1.mergeScan(function (acc, x) { return Observable.of(acc.concat(x)); }, []);
+
+ expectObservable(source).toBe(expected, values);
+ });
+
+ it('should handle errors in the projection function', function () {
+ var e1 = hot('--a--^--b--c--d--e--f--g--|');
+ var expected = '---u--v--#';
+
+ var values = {
+ u: ['b'],
+ v: ['b', 'c']
+ };
+
+ var source = e1.mergeScan(function (acc, x) {
+ if (x === 'd') {
+ throw 'bad!';
+ }
+ return Observable.of(acc.concat(x));
+ }, []);
+
+ expectObservable(source).toBe(expected, values, 'bad!');
+ });
+
+ it('handle empty', function () {
+ var e1 = Observable.empty();
+ var expected = '(u|)';
+
+ var values = {
+ u: []
+ };
+
+ var source = e1.mergeScan(function (acc, x) { return Observable.of(acc.concat(x)); }, []);
+
+ expectObservable(source).toBe(expected, values);
+ });
+
+ it('handle never', function () {
+ var e1 = Observable.never();
+ var expected = '-';
+
+ var source = e1.mergeScan(function (acc, x) { return Observable.of(acc.concat(x)); }, []);
+
+ expectObservable(source).toBe(expected);
+ });
+
+ it('handle throw', function () {
+ var e1 = Observable.throw('bad!');
+ var expected = '#';
+
+ var source = e1.mergeScan(function (acc, x) { return Observable.of(acc.concat(x)); }, []);
+
+ expectObservable(source).toBe(expected, undefined, 'bad!');
+ });
+
+ it('should mergeScan unsubscription', function () {
+ var e1 = hot('--a--^--b--c--d--e--f--g--|');
+ var expected = '---u--v--w--x--';
+ var sub = '^ !';
+ var values = {
+ u: ['b'],
+ v: ['b', 'c'],
+ w: ['b', 'c', 'd'],
+ x: ['b', 'c', 'd', 'e'],
+ y: ['b', 'c', 'd', 'e', 'f'],
+ z: ['b', 'c', 'd', 'e', 'f', 'g']
+ };
+
+ var source = e1.mergeScan(function (acc, x) { return Observable.of(acc.concat(x)); }, []);
+
+ expectObservable(source, sub).toBe(expected, values);
+ });
+});

src/Rx.KitchenSink.ts

@@ -12,6 +12,7 @@ interface KitchenSinkOperators<T> extends CoreOperators<T> {
  max?: <T, R>(comparer?: (x: R, y: T) => R) => Observable<R>;
  min?: <T, R>(comparer?: (x: R, y: T) => R) => Observable<R>;
  timeInterval?: <T>(scheduler?: IScheduler) => Observable<T>;
+ mergeScan?: <T, R>(project: (acc: R, x: T) => Observable<R>, seed: R) => Observable<R>;
 }
 
 // operators
@@ -196,6 +197,9 @@ import mergeMapTo from './operators/mergeMapTo';
 observableProto.mergeMapTo = mergeMapTo;
 observableProto.flatMapTo = mergeMapTo;
 
+import mergeScan from './operators/extended/mergeScan';
+observableProto.mergeScan = mergeScan;
+
 import min from './operators/extended/min';
 observableProto.min = min;
 

src/operators/extended/mergeScan.ts

@@ -0,0 +1,91 @@
+import Operator from '../../Operator';
+import Observable from '../../Observable';
+import Subscriber from '../../Subscriber';
+import Subscription from '../../Subscription';
+import tryCatch from '../../util/tryCatch';
+import { errorObject } from '../../util/errorObject';
+import subscribeToResult from '../../util/subscribeToResult';
+import OuterSubscriber from '../../OuterSubscriber';
+
+export default function mergeScan<T, R>(project: (acc: R, x: T) => Observable<R>, seed: R) {
+ return this.lift(new MergeScanOperator(project, seed));
+}
+
+export class MergeScanOperator<T, R> implements Operator<T, R> {
+ constructor(private project: (acc: R, x: T) => Observable<R>,
+ private seed: R,
+ private concurrent: number = Number.POSITIVE_INFINITY) {
+ }
+
+ call(subscriber: Subscriber<R>): Subscriber<T> {
+ return new MergeScanSubscriber(
+ subscriber, this.project, this.seed, this.concurrent
+ );
+ }
+}
+
+export class MergeScanSubscriber<T, R> extends OuterSubscriber<T, R> {
+ private hasValue: boolean = false;
+ private hasCompleted: boolean = false;
+ private buffer: Observable<any>[] = [];
+ private active: number = 0;
+ protected index: number = 0;
+
+ constructor(destination: Subscriber<R>,
+ private project: (acc: R, x: T) => Observable<R>,
+ private acc: R,
+ private concurrent: number = Number.POSITIVE_INFINITY) {
+ super(destination);
+ }
+
+ _next(value: any): void {
+ if (this.active < this.concurrent) {
+ const index = this.index++;
+ const ish = tryCatch(this.project)(this.acc, value);
+ const destination = this.destination;
+ if (ish === errorObject) {
+ destination.error(ish.e);
+ } else {
+ this.active++;
+ this._innerSub(ish, value, index);
+ }
+ } else {
+ this.buffer.push(value);
+ }
+ }
+
+ _innerSub(ish: any, value: T, index: number): void {
+ this.add(subscribeToResult<T, R>(this, ish, value, index));
+ }
+
+ _complete(): void {
+ this.hasCompleted = true;
+ if (this.active === 0 && this.buffer.length === 0) {
+ if (this.hasValue === false) {
+ this.destination.next(this.acc);
+ }
+ this.destination.complete();
+ }
+ }
+
+ notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+ const { destination } = this;
+ this.acc = innerValue;
+ this.hasValue = true;
+ destination.next(innerValue);
+ }
+
+ notifyComplete(innerSub: Subscription<T>): void {
+ const buffer = this.buffer;
+ this.remove(innerSub);
+ this.active--;
+ if (buffer.length > 0) {
+ this._next(buffer.shift());
+ } else if (this.active === 0 && this.hasCompleted) {
+ if (this.hasValue === false) {
+ this.destination.next(this.acc);
+ }
+ this.destination.complete();
+ }
+ }
+}