feat(mergeScan): Add index to the accumulator function (#4458)

* 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.

Author: martinsik

spec/operators/mergeScan-spec.ts

@@ -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]);
+  });
 });

src/internal/operators/mergeScan.ts

@@ -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) {
   }
@@ -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);
@@ -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);
       }