1import { Observable } from '../Observable'; 2import { Operator } from '../Operator'; 3import { Subscriber } from '../Subscriber'; 4import { Subscription } from '../Subscription'; 5import { OuterSubscriber } from '../OuterSubscriber'; 6import { InnerSubscriber } from '../InnerSubscriber'; 7import { subscribeToResult } from '../util/subscribeToResult'; 8import { MonoTypeOperatorFunction, OperatorFunction, ObservableInput, SchedulerLike } from '../types'; 9 10/* tslint:disable:max-line-length */ 11export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>, concurrent?: number, scheduler?: SchedulerLike): OperatorFunction<T, R>; 12export 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 */ 67export 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 75export 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 86interface 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 */ 98export 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} 181