feat(mergeScan): add index to the accumulator function

Closes ReactiveX#4441

spec-dtslint/operators/mergeScan-spec.ts

@@ -21,6 +21,10 @@ it('should support a currency', () => {
  const o = of(1, 2, 3).pipe(mergeScan((acc, value) => of(acc + value), '', 47)); // $ExpectType Observable<string>
 });
 
+it('should support an index parameter', () => {
+ const o = of(1, 2, 3).pipe(mergeScan((acc, value, index) => of(index), 0)); // $ExpectType Observable<number>
+});
+
 it('should enforce types', () => {
  const o = of(1, 2, 3).pipe(mergeScan()); // $ExpectError
 });

spec/operators/mergeScan-spec.ts

@@ -419,4 +419,17 @@ 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 expected = [0, 1, 2, 3];
+ 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(expected);
+ });
 });

src/internal/operators/mergeScan.ts

@@ -45,14 +45,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) {
  }
@@ -77,7 +77,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);
@@ -86,7 +86,7 @@ export class MergeScanSubscriber<T, R> extends OuterSubscriber<T, R> {
  protected _next(value: any): void {
  if (this.active < this.concurrent) {
  const index = this.index++;
- const ish = tryCatch(this.accumulator)(this.acc, value);
+ const ish = tryCatch(this.accumulator)(this.acc, value, index);
  const destination = this.destination;
  if (ish === errorObject) {
  destination.error(errorObject.e);