Skip to content

Commit

Permalink
feat(mergeScan): Add index to the accumulator function (#4458)
Browse files Browse the repository at this point in the history
* feat(mergeScan): add index to the accumulator function

Closes #4441

* test(mergeScan): use marbles instead of "of()"

* test(mergeScan): update spec, remove unnecessary dtest

* readd index lost in merge.
  • Loading branch information
martinsik authored and benlesh committed Jan 30, 2019
1 parent 0ee4fbe commit f5e143d
Show file tree
Hide file tree
Showing 2 changed files with 16 additions and 4 deletions.
12 changes: 12 additions & 0 deletions spec/operators/mergeScan-spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -419,4 +419,16 @@ describe('mergeScan', () => {
expectSubscriptions(inner[1].subscriptions).toBe(ysubs);
expectSubscriptions(inner[2].subscriptions).toBe(zsubs);
});

it('should pass current index to accumulator', () => {
const recorded: number[] = [];
const e1 = of('a', 'b', 'c', 'd');

e1.pipe(mergeScan((acc, x, index) => {
recorded.push(index);
return of(x);
}, 0)).subscribe();

expect(recorded).to.deep.equal([0, 1, 2, 3]);
});
});
8 changes: 4 additions & 4 deletions src/internal/operators/mergeScan.ts
Original file line number Diff line number Diff line change
Expand Up @@ -46,14 +46,14 @@ import { ObservableInput, OperatorFunction } from '../types';
* @method mergeScan
* @owner Observable
*/
export function mergeScan<T, R>(accumulator: (acc: R, value: T) => ObservableInput<R>,
export function mergeScan<T, R>(accumulator: (acc: R, value: T, index: number) => ObservableInput<R>,
seed: R,
concurrent: number = Number.POSITIVE_INFINITY): OperatorFunction<T, R> {
return (source: Observable<T>) => source.lift(new MergeScanOperator(accumulator, seed, concurrent));
}

export class MergeScanOperator<T, R> implements Operator<T, R> {
constructor(private accumulator: (acc: R, value: T) => ObservableInput<R>,
constructor(private accumulator: (acc: R, value: T, index: number) => ObservableInput<R>,
private seed: R,
private concurrent: number) {
}
Expand All @@ -78,7 +78,7 @@ export class MergeScanSubscriber<T, R> extends OuterSubscriber<T, R> {
protected index: number = 0;

constructor(destination: Subscriber<R>,
private accumulator: (acc: R, value: T) => ObservableInput<R>,
private accumulator: (acc: R, value: T, index: number) => ObservableInput<R>,
private acc: R,
private concurrent: number) {
super(destination);
Expand All @@ -91,7 +91,7 @@ export class MergeScanSubscriber<T, R> extends OuterSubscriber<T, R> {
let ish;
try {
const { accumulator } = this;
ish = accumulator(this.acc, value);
ish = accumulator(this.acc, value, index);
} catch (e) {
return destination.error(e);
}
Expand Down

0 comments on commit f5e143d

Please sign in to comment.