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