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

   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 { MonoTypeOperatorFunction, OperatorFunction, ObservableInput, SchedulerLike } from '../types';
   9
  10 /* tslint:disable:max-line-length */
  11 export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>, concurrent?: number, scheduler?: SchedulerLike): OperatorFunction<T, R>;
  12 export function expand<T>(project: (value: T, index: number) => ObservableInput<T>, concurrent?: number, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>;
  13 /* tslint:enable:max-line-length */
  14
  15 /**
  16  * Recursively projects each source value to an Observable which is merged in
  17  * the output Observable.
  18  *
  19  * <span class="informal">It's similar to {@link mergeMap}, but applies the
  20  * projection function to every source value as well as every output value.
  21  * It's recursive.</span>
  22  *
  23  * ![](expand.png)
  24  *
  25  * Returns an Observable that emits items based on applying a function that you
  26  * supply to each item emitted by the source Observable, where that function
  27  * returns an Observable, and then merging those resulting Observables and
  28  * emitting the results of this merger. *Expand* will re-emit on the output
  29  * Observable every source value. Then, each output value is given to the
  30  * `project` function which returns an inner Observable to be merged on the
  31  * output Observable. Those output values resulting from the projection are also
  32  * given to the `project` function to produce new output values. This is how
  33  * *expand* behaves recursively.
  34  *
  35  * ## Example
  36  * Start emitting the powers of two on every click, at most 10 of them
  37  * ```ts
  38  * import { fromEvent, of } from 'rxjs';
  39  * import { expand, mapTo, delay, take } from 'rxjs/operators';
  40  *
  41  * const clicks = fromEvent(document, 'click');
  42  * const powersOfTwo = clicks.pipe(
  43  *   mapTo(1),
  44  *   expand(x => of(2 * x).pipe(delay(1000))),
  45  *   take(10),
  46  * );
  47  * powersOfTwo.subscribe(x => console.log(x));
  48  * ```
  49  *
  50  * @see {@link mergeMap}
  51  * @see {@link mergeScan}
  52  *
  53  * @param {function(value: T, index: number) => Observable} project A function
  54  * that, when applied to an item emitted by the source or the output Observable,
  55  * returns an Observable.
  56  * @param {number} [concurrent=Number.POSITIVE_INFINITY] Maximum number of input
  57  * Observables being subscribed to concurrently.
  58  * @param {SchedulerLike} [scheduler=null] The {@link SchedulerLike} to use for subscribing to
  59  * each projected inner Observable.
  60  * @return {Observable} An Observable that emits the source values and also
  61  * result of applying the projection function to each value emitted on the
  62  * output Observable and and merging the results of the Observables obtained
  63  * from this transformation.
  64  * @method expand
  65  * @owner Observable
  66  */
  67 export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>,
  68                              concurrent: number = Number.POSITIVE_INFINITY,
  69                              scheduler: SchedulerLike = undefined): OperatorFunction<T, R> {
  70   concurrent = (concurrent || 0) < 1 ? Number.POSITIVE_INFINITY : concurrent;
  71
  72   return (source: Observable<T>) => source.lift(new ExpandOperator(project, concurrent, scheduler));
  73 }
  74
  75 export class ExpandOperator<T, R> implements Operator<T, R> {
  76   constructor(private project: (value: T, index: number) => ObservableInput<R>,
  77               private concurrent: number,
  78               private scheduler: SchedulerLike) {
  79   }
  80
  81   call(subscriber: Subscriber<R>, source: any): any {
  82     return source.subscribe(new ExpandSubscriber(subscriber, this.project, this.concurrent, this.scheduler));
  83   }
  84 }
  85
  86 interface DispatchArg<T, R> {
  87   subscriber: ExpandSubscriber<T, R>;
  88   result: ObservableInput<R>;
  89   value: any;
  90   index: number;
  91 }
  92
  93 /**
  94  * We need this JSDoc comment for affecting ESDoc.
  95  * @ignore
  96  * @extends {Ignored}
  97  */
  98 export class ExpandSubscriber<T, R> extends OuterSubscriber<T, R> {
  99   private index: number = 0;
 100   private active: number = 0;
 101   private hasCompleted: boolean = false;
 102   private buffer: any[];
 103
 104   constructor(destination: Subscriber<R>,
 105               private project: (value: T, index: number) => ObservableInput<R>,
 106               private concurrent: number,
 107               private scheduler: SchedulerLike) {
 108     super(destination);
 109     if (concurrent < Number.POSITIVE_INFINITY) {
 110       this.buffer = [];
 111     }
 112   }
 113
 114   private static dispatch<T, R>(arg: DispatchArg<T, R>): void {
 115     const {subscriber, result, value, index} = arg;
 116     subscriber.subscribeToProjection(result, value, index);
 117   }
 118
 119   protected _next(value: any): void {
 120     const destination = this.destination;
 121
 122     if (destination.closed) {
 123       this._complete();
 124       return;
 125     }
 126
 127     const index = this.index++;
 128     if (this.active < this.concurrent) {
 129       destination.next(value);
 130       try {
 131         const { project } = this;
 132         const result = project(value, index);
 133         if (!this.scheduler) {
 134           this.subscribeToProjection(result, value, index);
 135         } else {
 136           const state: DispatchArg<T, R> = { subscriber: this, result, value, index };
 137           const destination = this.destination as Subscription;
 138           destination.add(this.scheduler.schedule<DispatchArg<T, R>>(ExpandSubscriber.dispatch, 0, state));
 139         }
 140       } catch (e) {
 141         destination.error(e);
 142       }
 143     } else {
 144       this.buffer.push(value);
 145     }
 146   }
 147
 148   private subscribeToProjection(result: any, value: T, index: number): void {
 149     this.active++;
 150     const destination = this.destination as Subscription;
 151     destination.add(subscribeToResult<T, R>(this, result, value, index));
 152   }
 153
 154   protected _complete(): void {
 155     this.hasCompleted = true;
 156     if (this.hasCompleted && this.active === 0) {
 157       this.destination.complete();
 158     }
 159     this.unsubscribe();
 160   }
 161
 162   notifyNext(outerValue: T, innerValue: R,
 163              outerIndex: number, innerIndex: number,
 164              innerSub: InnerSubscriber<T, R>): void {
 165     this._next(innerValue);
 166   }
 167
 168   notifyComplete(innerSub: Subscription): void {
 169     const buffer = this.buffer;
 170     const destination = this.destination as Subscription;
 171     destination.remove(innerSub);
 172     this.active--;
 173     if (buffer && buffer.length > 0) {
 174       this._next(buffer.shift());
 175     }
 176     if (this.hasCompleted && this.active === 0) {
 177       this.destination.complete();
 178     }
 179   }
 180 }