.config/coc/extensions/node_modules/coc-prettier/node_modules/rxjs/src/internal/operators/mergeMap.ts

   1 import { Observable } from '../Observable';
   2 import { Operator } from '../Operator';
   3 import { Subscriber } from '../Subscriber';
   4 import { Subscription } from '../Subscription';
   5 import { ObservableInput, OperatorFunction, ObservedValueOf } from '../types';
   6 import { map } from './map';
   7 import { from } from '../observable/from';
   8 import { SimpleOuterSubscriber, SimpleInnerSubscriber, innerSubscribe } from '../innerSubscribe';
   9
  10 /* tslint:disable:max-line-length */
  11 export function mergeMap<T, O extends ObservableInput<any>>(project: (value: T, index: number) => O, concurrent?: number): OperatorFunction<T, ObservedValueOf<O>>;
  12 /** @deprecated resultSelector no longer supported, use inner map instead */
  13 export function mergeMap<T, O extends ObservableInput<any>>(project: (value: T, index: number) => O, resultSelector: undefined, concurrent?: number): OperatorFunction<T, ObservedValueOf<O>>;
  14 /** @deprecated resultSelector no longer supported, use inner map instead */
  15 export function mergeMap<T, R, O extends ObservableInput<any>>(project: (value: T, index: number) => O, resultSelector: (outerValue: T, innerValue: ObservedValueOf<O>, outerIndex: number, innerIndex: number) => R, concurrent?: number): OperatorFunction<T, R>;
  16 /* tslint:enable:max-line-length */
  17
  18 /**
  19  * Projects each source value to an Observable which is merged in the output
  20  * Observable.
  21  *
  22  * <span class="informal">Maps each value to an Observable, then flattens all of
  23  * these inner Observables using {@link mergeAll}.</span>
  24  *
  25  * ![](mergeMap.png)
  26  *
  27  * Returns an Observable that emits items based on applying a function that you
  28  * supply to each item emitted by the source Observable, where that function
  29  * returns an Observable, and then merging those resulting Observables and
  30  * emitting the results of this merger.
  31  *
  32  * ## Example
  33  * Map and flatten each letter to an Observable ticking every 1 second
  34  * ```ts
  35  * import { of, interval } from 'rxjs';
  36  * import { mergeMap, map } from 'rxjs/operators';
  37  *
  38  * const letters = of('a', 'b', 'c');
  39  * const result = letters.pipe(
  40  *   mergeMap(x => interval(1000).pipe(map(i => x+i))),
  41  * );
  42  * result.subscribe(x => console.log(x));
  43  *
  44  * // Results in the following:
  45  * // a0
  46  * // b0
  47  * // c0
  48  * // a1
  49  * // b1
  50  * // c1
  51  * // continues to list a,b,c with respective ascending integers
  52  * ```
  53  *
  54  * @see {@link concatMap}
  55  * @see {@link exhaustMap}
  56  * @see {@link merge}
  57  * @see {@link mergeAll}
  58  * @see {@link mergeMapTo}
  59  * @see {@link mergeScan}
  60  * @see {@link switchMap}
  61  *
  62  * @param {function(value: T, ?index: number): ObservableInput} project A function
  63  * that, when applied to an item emitted by the source Observable, returns an
  64  * Observable.
  65  * @param {number} [concurrent=Number.POSITIVE_INFINITY] Maximum number of input
  66  * Observables being subscribed to concurrently.
  67  * @return {Observable} An Observable that emits the result of applying the
  68  * projection function (and the optional deprecated `resultSelector`) to each item
  69  * emitted by the source Observable and merging the results of the Observables
  70  * obtained from this transformation.
  71  */
  72 export function mergeMap<T, R, O extends ObservableInput<any>>(
  73   project: (value: T, index: number) => O,
  74   resultSelector?: ((outerValue: T, innerValue: ObservedValueOf<O>, outerIndex: number, innerIndex: number) => R) | number,
  75   concurrent: number = Number.POSITIVE_INFINITY
  76 ): OperatorFunction<T, ObservedValueOf<O>|R> {
  77   if (typeof resultSelector === 'function') {
  78     // DEPRECATED PATH
  79     return (source: Observable<T>) => source.pipe(
  80       mergeMap((a, i) => from(project(a, i)).pipe(
  81         map((b: any, ii: number) => resultSelector(a, b, i, ii)),
  82       ), concurrent)
  83     );
  84   } else if (typeof resultSelector === 'number') {
  85     concurrent = resultSelector;
  86   }
  87   return (source: Observable<T>) => source.lift(new MergeMapOperator(project, concurrent));
  88 }
  89
  90 export class MergeMapOperator<T, R> implements Operator<T, R> {
  91   constructor(private project: (value: T, index: number) => ObservableInput<R>,
  92               private concurrent: number = Number.POSITIVE_INFINITY) {
  93   }
  94
  95   call(observer: Subscriber<R>, source: any): any {
  96     return source.subscribe(new MergeMapSubscriber(
  97       observer, this.project, this.concurrent
  98     ));
  99   }
 100 }
 101
 102 /**
 103  * We need this JSDoc comment for affecting ESDoc.
 104  * @ignore
 105  * @extends {Ignored}
 106  */
 107 export class MergeMapSubscriber<T, R> extends SimpleOuterSubscriber<T, R> {
 108   private hasCompleted: boolean = false;
 109   private buffer: T[] = [];
 110   private active: number = 0;
 111   protected index: number = 0;
 112
 113   constructor(destination: Subscriber<R>,
 114               private project: (value: T, index: number) => ObservableInput<R>,
 115               private concurrent: number = Number.POSITIVE_INFINITY) {
 116     super(destination);
 117   }
 118
 119   protected _next(value: T): void {
 120     if (this.active < this.concurrent) {
 121       this._tryNext(value);
 122     } else {
 123       this.buffer.push(value);
 124     }
 125   }
 126
 127   protected _tryNext(value: T) {
 128     let result: ObservableInput<R>;
 129     const index = this.index++;
 130     try {
 131       result = this.project(value, index);
 132     } catch (err) {
 133       this.destination.error!(err);
 134       return;
 135     }
 136     this.active++;
 137     this._innerSub(result);
 138   }
 139
 140   private _innerSub(ish: ObservableInput<R>): void {
 141     const innerSubscriber = new SimpleInnerSubscriber(this);
 142     const destination = this.destination as Subscription;
 143     destination.add(innerSubscriber);
 144     const innerSubscription = innerSubscribe(ish, innerSubscriber);
 145     // The returned subscription will usually be the subscriber that was
 146     // passed. However, interop subscribers will be wrapped and for
 147     // unsubscriptions to chain correctly, the wrapper needs to be added, too.
 148     if (innerSubscription !== innerSubscriber) {
 149       destination.add(innerSubscription);
 150     }
 151   }
 152
 153   protected _complete(): void {
 154     this.hasCompleted = true;
 155     if (this.active === 0 && this.buffer.length === 0) {
 156       this.destination.complete!();
 157     }
 158     this.unsubscribe();
 159   }
 160
 161   notifyNext(innerValue: R): void {
 162     this.destination.next!(innerValue);
 163   }
 164
 165   notifyComplete(): void {
 166     const buffer = this.buffer;
 167     this.active--;
 168     if (buffer.length > 0) {
 169       this._next(buffer.shift()!);
 170     } else if (this.active === 0 && this.hasCompleted) {
 171       this.destination.complete!();
 172     }
 173   }
 174 }
 175
 176 /**
 177  * @deprecated renamed. Use {@link mergeMap}
 178  */
 179 export const flatMap = mergeMap;