fix(mergeMap): allow concurrent to be set as the second argument for …

…mergeMap and mergeMapTo This behavior is similar to how merge works
ReactiveX · Mar 22, 2016 · c003468 · c003468
1 parent fa43a22
commit c003468
Show file tree

Hide file tree

Showing 4 changed files with 155 additions and 4 deletions.
diff --git a/spec/operators/mergeMap-spec.ts b/spec/operators/mergeMap-spec.ts
@@ -1,6 +1,6 @@
 import * as Rx from '../../dist/cjs/Rx';
 declare const {hot, cold, expectObservable, expectSubscriptions};
-import {DoneSignature} from '../helpers/test-helper';
+import {DoneSignature, type} from '../helpers/test-helper';
 
 const Observable = Rx.Observable;
 
@@ -352,6 +352,88 @@ describe('Observable.prototype.mergeMap', () => {
     expectSubscriptions(hotC.subscriptions).toBe(csubs);
   });
 
+  it('should mergeMap to many cold Observable, with parameter concurrency=1, without resultSelector', () => {
+    const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
+    const e1 =     hot('-a-------b-------c---|                                        ');
+    const e1subs =     '^                                                            !';
+    const inner =  cold('----i---j---k---l---|                                        ', values);
+    const innersubs = [' ^                   !                                        ',
+                     '                     ^                   !                    ',
+                     '                                         ^                   !'];
+    const expected =   '-----i---j---k---l-------i---j---k---l-------i---j---k---l---|';
+
+    function project() { return inner; }
+    const result = e1.mergeMap(project, 1);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(inner.subscriptions).toBe(innersubs);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
+
+  it('should mergeMap to many cold Observable, with parameter concurrency=2, without resultSelector', () => {
+    const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
+    const e1 =     hot('-a-------b-------c---|                    ');
+    const e1subs =     '^                                        !';
+    const inner =  cold('----i---j---k---l---|                    ', values);
+    const innersubs = [' ^                   !                    ',
+                     '         ^                   !            ',
+                     '                     ^                   !'];
+    const expected =   '-----i---j---(ki)(lj)k---(li)j---k---l---|';
+
+    function project() { return inner; }
+    const result = e1.mergeMap(project, 2);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(inner.subscriptions).toBe(innersubs);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
+
+  it('should mergeMap to many hot Observable, with parameter concurrency=1, without resultSelector', () => {
+    const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
+    const e1 =     hot('-a-------b-------c---|                                        ');
+    const e1subs =     '^                                                            !';
+    const hotA =   hot('x----i---j---k---l---|                                        ', values);
+    const hotB =   hot('-x-x-xxxx-x-x-xxxxx-x----i---j---k---l---|                    ', values);
+    const hotC =   hot('x-xxxx---x-x-x-x-x-xx--x--x-x--x--xxxx-x-----i---j---k---l---|', values);
+    const asubs =      ' ^                   !                                        ';
+    const bsubs =      '                     ^                   !                    ';
+    const csubs =      '                                         ^                   !';
+    const expected =   '-----i---j---k---l-------i---j---k---l-------i---j---k---l---|';
+    const inners = { a: hotA, b: hotB, c: hotC };
+
+    function project(x) { return inners[x]; }
+    const result = e1.mergeMap(project, 1);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(hotA.subscriptions).toBe(asubs);
+    expectSubscriptions(hotB.subscriptions).toBe(bsubs);
+    expectSubscriptions(hotC.subscriptions).toBe(csubs);
+  });
+
+  it('should mergeMap to many hot Observable, with parameter concurrency=2, without resultSelector', () => {
+    const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
+    const e1 =     hot('-a-------b-------c---|                    ');
+    const e1subs =     '^                                        !';
+    const hotA =   hot('x----i---j---k---l---|                    ', values);
+    const hotB =   hot('-x-x-xxxx----i---j---k---l---|            ', values);
+    const hotC =   hot('x-xxxx---x-x-x-x-x-xx----i---j---k---l---|', values);
+    const asubs =      ' ^                   !                    ';
+    const bsubs =      '         ^                   !            ';
+    const csubs =      '                     ^                   !';
+    const expected =   '-----i---j---(ki)(lj)k---(li)j---k---l---|';
+    const inners = { a: hotA, b: hotB, c: hotC };
+
+    function project(x) { return inners[x]; }
+    const result = e1.mergeMap(project, 2);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(hotA.subscriptions).toBe(asubs);
+    expectSubscriptions(hotB.subscriptions).toBe(bsubs);
+    expectSubscriptions(hotC.subscriptions).toBe(csubs);
+  });
+
   it('should mergeMap many complex, where all inners are finite', () => {
     const a =   cold( '-#'                                                  );
     const b =   cold(   '-#'                                                );
@@ -663,4 +745,17 @@ describe('Observable.prototype.mergeMap', () => {
 
     expect(completed).toBe(true);
   });
+
+  it('should support type signatures', () => {
+    type(() => {
+      let o: Rx.Observable<number>;
+
+      /* tslint:disable:no-unused-variable */
+      let a1: Rx.Observable<string> = o.mergeMap(x => x.toString());
+      let a2: Rx.Observable<string> = o.mergeMap(x => x.toString(), 3);
+      let a3: Rx.Observable<{ o: number; i: string; }> = o.mergeMap(x => x.toString(), (o, i) => ({ o, i }));
+      let a4: Rx.Observable<{ o: number; i: string; }> = o.mergeMap(x => x.toString(), (o, i) => ({ o, i }), 3);
+      /* tslint:enable:no-unused-variable */
+    });
+  });
 });
diff --git a/spec/operators/mergeMapTo-spec.ts b/spec/operators/mergeMapTo-spec.ts
@@ -1,6 +1,6 @@
 import * as Rx from '../../dist/cjs/Rx';
 declare const {hot, cold, expectObservable, expectSubscriptions};
-import {DoneSignature} from '../helpers/test-helper';
+import {DoneSignature, type} from '../helpers/test-helper';
 
 const Observable = Rx.Observable;
 
@@ -284,6 +284,40 @@ describe('Observable.prototype.mergeMapTo', () => {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
+  it('should mergeMapTo many cold Observable, with parameter concurrency=1, without resultSelector', () => {
+    const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
+    const e1 =     hot('-a-------b-------c---|                                        ');
+    const e1subs =     '^                                                            !';
+    const inner =  cold('----i---j---k---l---|                                        ', values);
+    const innersubs = [' ^                   !                                        ',
+                     '                     ^                   !                    ',
+                     '                                         ^                   !'];
+    const expected =   '-----i---j---k---l-------i---j---k---l-------i---j---k---l---|';
+
+    const result = e1.mergeMapTo(inner, 1);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(inner.subscriptions).toBe(innersubs);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
+
+  it('should mergeMap to many cold Observable, with parameter concurrency=2, without resultSelector', () => {
+    const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
+    const e1 =     hot('-a-------b-------c---|                    ');
+    const e1subs =     '^                                        !';
+    const inner =  cold('----i---j---k---l---|                    ', values);
+    const innersubs = [' ^                   !                    ',
+                     '         ^                   !            ',
+                     '                     ^                   !'];
+    const expected =   '-----i---j---(ki)(lj)k---(li)j---k---l---|';
+
+    const result = e1.mergeMapTo(inner, 2);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(inner.subscriptions).toBe(innersubs);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
+
   it('should mergeMapTo many outer to arrays', () => {
     const e1 =   hot('2-----4--------3--------2-------|');
     const e1subs =   '^                               !';
@@ -402,4 +436,18 @@ describe('Observable.prototype.mergeMapTo', () => {
 
     expect(completed).toBe(true);
   });
+
+  it('should support type signatures', () => {
+    type(() => {
+      let o: Rx.Observable<number>;
+      let m: Rx.Observable<string>;
+
+      /* tslint:disable:no-unused-variable */
+      let a1: Rx.Observable<string> = o.mergeMapTo(m);
+      let a2: Rx.Observable<string> = o.mergeMapTo(m, 3);
+      let a3: Rx.Observable<{ o: number; i: string; }> = o.mergeMapTo(m, (o, i) => ({ o, i }));
+      let a4: Rx.Observable<{ o: number; i: string; }> = o.mergeMapTo(m, (o, i) => ({ o, i }), 3);
+      /* tslint:enable:no-unused-variable */
+    });
+  });
 });
diff --git a/src/operator/mergeMap.ts b/src/operator/mergeMap.ts
@@ -20,8 +20,12 @@ import {InnerSubscriber} from '../InnerSubscriber';
  * @owner Observable
  */
 export function mergeMap<T, I, R>(project: (value: T, index: number) => ObservableInput<I>,
-                                  resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R | number,
+                                  resultSelector?: ((outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) | number,
                                   concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
+  if (typeof resultSelector === 'number') {
+    concurrent = <number>resultSelector;
+    resultSelector = null;
+  }
   return this.lift(new MergeMapOperator(project, <any>resultSelector, concurrent));
 }
 

diff --git a/src/operator/mergeMapTo.ts b/src/operator/mergeMapTo.ts
@@ -16,8 +16,12 @@ import {subscribeToResult} from '../util/subscribeToResult';
  * @owner Observable
  */
 export function mergeMapTo<T, I, R>(observable: Observable<I>,
-                                    resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R | number,
+                                    resultSelector?: ((outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) | number,
                                     concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
+  if (typeof resultSelector === 'number') {
+    concurrent = <number>resultSelector;
+    resultSelector = null;
+  }
   return this.lift(new MergeMapToOperator(observable, <any>resultSelector, concurrent));
 }