Skip to content

Commit 582c7be

Browse files
benleshbenlesh
committed
feat(mergeMapTo): simplify interface
- Implements mergeMapTo in terms of mergeMap - Removes resultSelector argument BREAKING CHANGE: `mergeMapTo` no longer accepts a resultSelector, to get this functionality, you'll want to use `mergeMap` and `map` together: `source.pipe(mergeMap(() => inner).pipe(map(y => x + y)))`
1 parent d293245 commit 582c7be

File tree

3 files changed

+11
-292
lines changed

3 files changed

+11
-292
lines changed

‎spec/operators/mergeMapTo-spec.ts‎

Lines changed: 0 additions & 141 deletions
Original file line numberDiff line numberDiff line change
@@ -75,56 +75,6 @@ describe('Observable.prototype.mergeMapTo', () => {
7575
});
7676
});
7777

78-
it('should mergeMapTo values to resolved promises with resultSelector', (done) => {
79-
const source = Rx.Observable.from([4, 3, 2, 1]);
80-
const resultSelectorCalledWith: number[][] = [];
81-
const inner = Observable.from(Promise.resolve(42));
82-
const resultSelector = function (outerVal: number, innerVal: number, outerIndex: number, innerIndex: number) {
83-
resultSelectorCalledWith.push([].slice.call(arguments));
84-
return 8;
85-
};
86-
87-
const results: number[] = [];
88-
const expectedCalls = [
89-
[4, 42, 0, 0],
90-
[3, 42, 1, 0],
91-
[2, 42, 2, 0],
92-
[1, 42, 3, 0],
93-
];
94-
source.mergeMapTo(inner, resultSelector).subscribe(
95-
(x) => {
96-
results.push(x);
97-
},
98-
(err) => {
99-
done(new Error('Subscriber error handler not supposed to be called.'));
100-
},
101-
() => {
102-
expect(results).to.deep.equal([8, 8, 8, 8]);
103-
expect(resultSelectorCalledWith).to.deep.equal(expectedCalls);
104-
done();
105-
});
106-
});
107-
108-
it('should mergeMapTo values to rejected promises with resultSelector', (done) => {
109-
const source = Rx.Observable.from([4, 3, 2, 1]);
110-
const inner = Observable.from(Promise.reject(42));
111-
const resultSelector = () => {
112-
throw 'this should not be called';
113-
};
114-
115-
source.mergeMapTo(inner, resultSelector).subscribe(
116-
(x) => {
117-
done(new Error('Subscriber next handler not supposed to be called.'));
118-
},
119-
(err) => {
120-
expect(err).to.equal(42);
121-
done();
122-
},
123-
() => {
124-
done(new Error('Subscriber complete handler not supposed to be called.'));
125-
});
126-
});
127-
12878
it('should mergeMapTo many outer values to many inner values', () => {
12979
const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
13080
const e1 = hot('-a-------b-------c-------d-------| ');
@@ -264,42 +214,6 @@ describe('Observable.prototype.mergeMapTo', () => {
264214
expectSubscriptions(e1.subscriptions).toBe(e1subs);
265215
});
266216

267-
it('should mergeMapTo many cold Observable, with parameter concurrency=1', () => {
268-
const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
269-
const e1 = hot('-a-------b-------c---| ');
270-
const e1subs = '^ !';
271-
const inner = cold('----i---j---k---l---| ', values);
272-
const innersubs = [' ^ ! ',
273-
' ^ ! ',
274-
' ^ !'];
275-
const expected = '-----i---j---k---l-------i---j---k---l-------i---j---k---l---|';
276-
277-
function resultSelector(oV: string, iV: string, oI: number, iI: number) { return iV; }
278-
const result = e1.mergeMapTo(inner, resultSelector, 1);
279-
280-
expectObservable(result).toBe(expected, values);
281-
expectSubscriptions(inner.subscriptions).toBe(innersubs);
282-
expectSubscriptions(e1.subscriptions).toBe(e1subs);
283-
});
284-
285-
it('should mergeMap to many cold Observable, with parameter concurrency=2', () => {
286-
const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
287-
const e1 = hot('-a-------b-------c---| ');
288-
const e1subs = '^ !';
289-
const inner = cold('----i---j---k---l---| ', values);
290-
const innersubs = [' ^ ! ',
291-
' ^ ! ',
292-
' ^ !'];
293-
const expected = '-----i---j---(ki)(lj)k---(li)j---k---l---|';
294-
295-
function resultSelector(oV: string, iV: string, oI: number, iI: number) { return iV; }
296-
const result = e1.mergeMapTo(inner, resultSelector, 2);
297-
298-
expectObservable(result).toBe(expected, values);
299-
expectSubscriptions(inner.subscriptions).toBe(innersubs);
300-
expectSubscriptions(e1.subscriptions).toBe(e1subs);
301-
});
302-
303217
it('should mergeMapTo many cold Observable, with parameter concurrency=1, without resultSelector', () => {
304218
const values = {i: 'foo', j: 'bar', k: 'baz', l: 'qux'};
305219
const e1 = hot('-a-------b-------c---| ');
@@ -345,18 +259,6 @@ describe('Observable.prototype.mergeMapTo', () => {
345259
expectSubscriptions(e1.subscriptions).toBe(e1subs);
346260
});
347261

348-
it('should mergeMapTo many outer to inner arrays, using resultSelector', () => {
349-
const e1 = hot('2-----4--------3--------2-------|');
350-
const e1subs = '^ !';
351-
const expected = '(2345)(4567)---(3456)---(2345)--|';
352-
353-
const source = e1.mergeMapTo(['0', '1', '2', '3'],
354-
(x, y) => String(parseInt(x) + parseInt(y)));
355-
356-
expectObservable(source).toBe(expected);
357-
expectSubscriptions(e1.subscriptions).toBe(e1subs);
358-
});
359-
360262
it('should mergeMapTo many outer to inner arrays, and outer throws', () => {
361263
const e1 = hot('2-----4--------3--------2-------#');
362264
const e1subs = '^ !';
@@ -368,18 +270,6 @@ describe('Observable.prototype.mergeMapTo', () => {
368270
expectSubscriptions(e1.subscriptions).toBe(e1subs);
369271
});
370272

371-
it('should mergeMapTo many outer to inner arrays, resultSelector, outer throws', () => {
372-
const e1 = hot('2-----4--------3--------2-------#');
373-
const e1subs = '^ !';
374-
const expected = '(2345)(4567)---(3456)---(2345)--#';
375-
376-
const source = e1.mergeMapTo(['0', '1', '2', '3'],
377-
(x, y) => String(parseInt(x) + parseInt(y)));
378-
379-
expectObservable(source).toBe(expected);
380-
expectSubscriptions(e1.subscriptions).toBe(e1subs);
381-
});
382-
383273
it('should mergeMapTo many outer to inner arrays, outer gets unsubscribed', () => {
384274
const e1 = hot('2-----4--------3--------2-------|');
385275
const e1subs = '^ !';
@@ -392,35 +282,6 @@ describe('Observable.prototype.mergeMapTo', () => {
392282
expectSubscriptions(e1.subscriptions).toBe(e1subs);
393283
});
394284

395-
it('should mergeMapTo many outer to inner arrays, resultSelector, outer unsubscribed', () => {
396-
const e1 = hot('2-----4--------3--------2-------|');
397-
const e1subs = '^ !';
398-
const unsub = ' !';
399-
const expected = '(2345)(4567)--';
400-
401-
const source = e1.mergeMapTo(['0', '1', '2', '3'],
402-
(x, y) => String(parseInt(x) + parseInt(y)));
403-
404-
expectObservable(source, unsub).toBe(expected);
405-
expectSubscriptions(e1.subscriptions).toBe(e1subs);
406-
});
407-
408-
it('should mergeMapTo many outer to inner arrays, resultSelector throws', () => {
409-
const e1 = hot('2-----4--------3--------2-------|');
410-
const e1subs = '^ !';
411-
const expected = '(2345)(4567)---#';
412-
413-
const source = e1.mergeMapTo(['0', '1', '2', '3'], (outer, inner) => {
414-
if (outer === '3') {
415-
throw 'error';
416-
}
417-
return String(parseInt(outer) + parseInt(inner));
418-
});
419-
420-
expectObservable(source).toBe(expected);
421-
expectSubscriptions(e1.subscriptions).toBe(e1subs);
422-
});
423-
424285
it('should map and flatten', () => {
425286
const source = Observable.of(1, 2, 3, 4).mergeMapTo(Observable.of('!'));
426287

@@ -460,8 +321,6 @@ describe('Observable.prototype.mergeMapTo', () => {
460321
/* tslint:disable:no-unused-variable */
461322
let a1: Rx.Observable<string> = o.mergeMapTo(m);
462323
let a2: Rx.Observable<string> = o.mergeMapTo(m, 3);
463-
let a3: Rx.Observable<{ o: number; i: string; }> = o.mergeMapTo(m, (o, i) => ({ o, i }));
464-
let a4: Rx.Observable<{ o: number; i: string; }> = o.mergeMapTo(m, (o, i) => ({ o, i }), 3);
465324
/* tslint:enable:no-unused-variable */
466325
});
467326
});

‎src/internal/operators/mergeMapTo.ts‎

Lines changed: 7 additions & 132 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,6 @@
1-
import { Observable } from '../Observable';
2-
import { Operator } from '../Operator';
3-
import { Subscriber } from '../Subscriber';
4-
import { Subscription } from '../Subscription';
5-
import { OuterSubscriber } from '../OuterSubscriber';
6-
import { InnerSubscriber } from '../InnerSubscriber';
7-
import { subscribeToResult } from '../util/subscribeToResult';
8-
import { ObservableInput, OperatorFunction, PartialObserver } from '../types';
9-
10-
/* tslint:disable:max-line-length */
11-
export function mergeMapTo<T, R>(observable: ObservableInput<R>, concurrent?: number): OperatorFunction<T, R>;
12-
export function mergeMapTo<T, I, R>(observable: ObservableInput<I>, resultSelector: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R, concurrent?: number): OperatorFunction<T, R>;
13-
/* tslint:enable:max-line-length */
1+
import { Observable, ObservableInput } from '../Observable';
2+
import { OperatorFunction } from '../../internal/types';
3+
import { mergeMap } from './mergeMap';
144

155
/**
166
* Projects each source value to the same Observable which is merged multiple
@@ -39,129 +29,14 @@ export function mergeMapTo<T, I, R>(observable: ObservableInput<I>, resultSelect
3929
*
4030
* @param {ObservableInput} innerObservable An Observable to replace each value from
4131
* the source Observable.
42-
* @param {function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any} [resultSelector]
43-
* A function to produce the value on the output Observable based on the values
44-
* and the indices of the source (outer) emission and the inner Observable
45-
* emission. The arguments passed to this function are:
46-
* - `outerValue`: the value that came from the source
47-
* - `innerValue`: the value that came from the projected Observable
48-
* - `outerIndex`: the "index" of the value that came from the source
49-
* - `innerIndex`: the "index" of the value from the projected Observable
5032
* @param {number} [concurrent=Number.POSITIVE_INFINITY] Maximum number of input
5133
* Observables being subscribed to concurrently.
5234
* @return {Observable} An Observable that emits items from the given
53-
* `innerObservable` (and optionally transformed through `resultSelector`) every
54-
* time a value is emitted on the source Observable.
35+
* `innerObservable`
5536
* @method mergeMapTo
5637
* @owner Observable
5738
*/
58-
export function mergeMapTo<T, I, R>(innerObservable: ObservableInput<I>,
59-
resultSelector?: ((outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) | number,
60-
concurrent: number = Number.POSITIVE_INFINITY): OperatorFunction<T, R> {
61-
if (typeof resultSelector === 'number') {
62-
concurrent = <number>resultSelector;
63-
resultSelector = null;
64-
}
65-
return (source: Observable<T>) => source.lift(new MergeMapToOperator(innerObservable, <any>resultSelector, concurrent));
66-
}
67-
68-
// TODO: Figure out correct signature here: an Operator<Observable<T>, R>
69-
// needs to implement call(observer: Subscriber<R>): Subscriber<Observable<T>>
70-
export class MergeMapToOperator<T, I, R> implements Operator<Observable<T>, R> {
71-
constructor(private ish: ObservableInput<I>,
72-
private resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
73-
private concurrent: number = Number.POSITIVE_INFINITY) {
74-
}
75-
76-
call(observer: Subscriber<R>, source: any): any {
77-
return source.subscribe(new MergeMapToSubscriber(observer, this.ish, this.resultSelector, this.concurrent));
78-
}
79-
}
80-
81-
/**
82-
* We need this JSDoc comment for affecting ESDoc.
83-
* @ignore
84-
* @extends {Ignored}
85-
*/
86-
export class MergeMapToSubscriber<T, I, R> extends OuterSubscriber<T, I> {
87-
private hasCompleted: boolean = false;
88-
private buffer: T[] = [];
89-
private active: number = 0;
90-
protected index: number = 0;
91-
92-
constructor(destination: Subscriber<R>,
93-
private ish: ObservableInput<I>,
94-
private resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
95-
private concurrent: number = Number.POSITIVE_INFINITY) {
96-
super(destination);
97-
}
98-
99-
protected _next(value: T): void {
100-
if (this.active < this.concurrent) {
101-
const resultSelector = this.resultSelector;
102-
const index = this.index++;
103-
const ish = this.ish;
104-
const destination = this.destination;
105-
106-
this.active++;
107-
this._innerSub(ish, destination, resultSelector, value, index);
108-
} else {
109-
this.buffer.push(value);
110-
}
111-
}
112-
113-
private _innerSub(ish: ObservableInput<I>,
114-
destination: PartialObserver<I>,
115-
resultSelector: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
116-
value: T,
117-
index: number): void {
118-
this.add(subscribeToResult<T, I>(this, ish, value, index));
119-
}
120-
121-
protected _complete(): void {
122-
this.hasCompleted = true;
123-
if (this.active === 0 && this.buffer.length === 0) {
124-
this.destination.complete();
125-
}
126-
}
127-
128-
notifyNext(outerValue: T, innerValue: I,
129-
outerIndex: number, innerIndex: number,
130-
innerSub: InnerSubscriber<T, I>): void {
131-
const { resultSelector, destination } = this;
132-
if (resultSelector) {
133-
this.trySelectResult(outerValue, innerValue, outerIndex, innerIndex);
134-
} else {
135-
destination.next(innerValue);
136-
}
137-
}
138-
139-
private trySelectResult(outerValue: T, innerValue: I,
140-
outerIndex: number, innerIndex: number): void {
141-
const { resultSelector, destination } = this;
142-
let result: R;
143-
try {
144-
result = resultSelector(outerValue, innerValue, outerIndex, innerIndex);
145-
} catch (err) {
146-
destination.error(err);
147-
return;
148-
}
149-
150-
destination.next(result);
151-
}
152-
153-
notifyError(err: any): void {
154-
this.destination.error(err);
155-
}
156-
157-
notifyComplete(innerSub: Subscription): void {
158-
const buffer = this.buffer;
159-
this.remove(innerSub);
160-
this.active--;
161-
if (buffer.length > 0) {
162-
this._next(buffer.shift());
163-
} else if (this.active === 0 && this.hasCompleted) {
164-
this.destination.complete();
165-
}
166-
}
39+
export function mergeMapTo<T, R>(innerObservable: ObservableInput<R>,
40+
concurrent: number = Number.POSITIVE_INFINITY): OperatorFunction<T, R> {
41+
return mergeMap(() => innerObservable, concurrent);
16742
}

‎src/internal/patching/operator/mergeMapTo.ts‎

Lines changed: 4 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,6 @@ import { Observable } from '../../Observable';
22
import { ObservableInput } from '../../types';
33
import { mergeMapTo as higherOrder } from '../../operators/mergeMapTo';
44

5-
/* tslint:disable:max-line-length */
6-
export function mergeMapTo<T, R>(this: Observable<T>, observable: ObservableInput<R>, concurrent?: number): Observable<R>;
7-
export function mergeMapTo<T, I, R>(this: Observable<T>, observable: ObservableInput<I>, resultSelector: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R, concurrent?: number): Observable<R>;
8-
/* tslint:enable:max-line-length */
9-
105
/**
116
* Projects each source value to the same Observable which is merged multiple
127
* times in the output Observable.
@@ -34,24 +29,14 @@ export function mergeMapTo<T, I, R>(this: Observable<T>, observable: ObservableI
3429
*
3530
* @param {ObservableInput} innerObservable An Observable to replace each value from
3631
* the source Observable.
37-
* @param {function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any} [resultSelector]
38-
* A function to produce the value on the output Observable based on the values
39-
* and the indices of the source (outer) emission and the inner Observable
40-
* emission. The arguments passed to this function are:
41-
* - `outerValue`: the value that came from the source
42-
* - `innerValue`: the value that came from the projected Observable
43-
* - `outerIndex`: the "index" of the value that came from the source
44-
* - `innerIndex`: the "index" of the value from the projected Observable
4532
* @param {number} [concurrent=Number.POSITIVE_INFINITY] Maximum number of input
4633
* Observables being subscribed to concurrently.
4734
* @return {Observable} An Observable that emits items from the given
48-
* `innerObservable` (and optionally transformed through `resultSelector`) every
49-
* time a value is emitted on the source Observable.
35+
* `innerObservable`.
5036
* @method mergeMapTo
5137
* @owner Observable
5238
*/
53-
export function mergeMapTo<T, I, R>(this: Observable<T>, innerObservable: ObservableInput<I>,
54-
resultSelector?: ((outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) | number,
55-
concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
56-
return higherOrder(innerObservable, resultSelector as any, concurrent)(this) as Observable<R>;
39+
export function mergeMapTo<T, R>(this: Observable<T>, innerObservable: ObservableInput<R>,
40+
concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
41+
return higherOrder(innerObservable, concurrent)(this) as Observable<R>;
5742
}

0 commit comments

Comments
 (0)