import { ObservableInput, OperatorFunction, ObservedValueOf } from '../types'; import { map } from './map'; import { innerFrom } from '../observable/innerFrom'; import { operate } from '../util/lift'; import { mergeInternals } from './mergeInternals'; import { isFunction } from '../util/isFunction'; /* tslint:disable:max-line-length */ export function mergeMap>( project: (value: T, index: number) => O, concurrent?: number ): OperatorFunction>; /** @deprecated The `resultSelector` parameter will be removed in v8. Use an inner `map` instead. Details: https://rxjs.dev/deprecations/resultSelector */ export function mergeMap>( project: (value: T, index: number) => O, resultSelector: undefined, concurrent?: number ): OperatorFunction>; /** @deprecated The `resultSelector` parameter will be removed in v8. Use an inner `map` instead. Details: https://rxjs.dev/deprecations/resultSelector */ export function mergeMap>( project: (value: T, index: number) => O, resultSelector: (outerValue: T, innerValue: ObservedValueOf, outerIndex: number, innerIndex: number) => R, concurrent?: number ): OperatorFunction; /* tslint:enable:max-line-length */ /** * Projects each source value to an Observable which is merged in the output * Observable. * * Maps each value to an Observable, then flattens all of * these inner Observables using {@link mergeAll}. * * ![](mergeMap.png) * * Returns an Observable that emits items based on applying a function that you * supply to each item emitted by the source Observable, where that function * returns an Observable, and then merging those resulting Observables and * emitting the results of this merger. * * ## Example * * Map and flatten each letter to an Observable ticking every 1 second * * ```ts * import { of, mergeMap, interval, map } from 'rxjs'; * * const letters = of('a', 'b', 'c'); * const result = letters.pipe( * mergeMap(x => interval(1000).pipe(map(i => x + i))) * ); * * result.subscribe(x => console.log(x)); * * // Results in the following: * // a0 * // b0 * // c0 * // a1 * // b1 * // c1 * // continues to list a, b, c every second with respective ascending integers * ``` * * @see {@link concatMap} * @see {@link exhaustMap} * @see {@link merge} * @see {@link mergeAll} * @see {@link mergeMapTo} * @see {@link mergeScan} * @see {@link switchMap} * * @param {function(value: T, ?index: number): ObservableInput} project A function * that, when applied to an item emitted by the source Observable, returns an * Observable. * @param {number} [concurrent=Infinity] Maximum number of input * Observables being subscribed to concurrently. * @return A function that returns an Observable that emits the result of * applying the projection function (and the optional deprecated * `resultSelector`) to each item emitted by the source Observable and merging * the results of the Observables obtained from this transformation. */ export function mergeMap>( project: (value: T, index: number) => O, resultSelector?: ((outerValue: T, innerValue: ObservedValueOf, outerIndex: number, innerIndex: number) => R) | number, concurrent: number = Infinity ): OperatorFunction | R> { if (isFunction(resultSelector)) { // DEPRECATED PATH return mergeMap((a, i) => map((b: any, ii: number) => resultSelector(a, b, i, ii))(innerFrom(project(a, i))), concurrent); } else if (typeof resultSelector === 'number') { concurrent = resultSelector; } return operate((source, subscriber) => mergeInternals(source, subscriber, project, concurrent)); }