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';
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 */
14 * Recursively projects each source value to an Observable which is merged in
15 * the output Observable.
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>
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.
34 * Start emitting the powers of two on every click, at most 10 of them
36 * import { fromEvent, of } from 'rxjs';
37 * import { expand, mapTo, delay, take } from 'rxjs/operators';
39 * const clicks = fromEvent(document, 'click');
40 * const powersOfTwo = clicks.pipe(
42 * expand(x => of(2 * x).pipe(delay(1000))),
45 * powersOfTwo.subscribe(x => console.log(x));
48 * @see {@link mergeMap}
49 * @see {@link mergeScan}
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.
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;
70 return (source: Observable<T>) => source.lift(new ExpandOperator(project, concurrent, scheduler));
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) {
79 call(subscriber: Subscriber<R>, source: any): any {
80 return source.subscribe(new ExpandSubscriber(subscriber, this.project, this.concurrent, this.scheduler));
84 interface DispatchArg<T, R> {
85 subscriber: ExpandSubscriber<T, R>;
86 result: ObservableInput<R>;
92 * We need this JSDoc comment for affecting ESDoc.
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[];
102 constructor(destination: Subscriber<R>,
103 private project: (value: T, index: number) => ObservableInput<R>,
104 private concurrent: number,
105 private scheduler?: SchedulerLike) {
107 if (concurrent < Number.POSITIVE_INFINITY) {
112 private static dispatch<T, R>(arg: DispatchArg<T, R>): void {
113 const {subscriber, result, value, index} = arg;
114 subscriber.subscribeToProjection(result, value, index);
117 protected _next(value: any): void {
118 const destination = this.destination;
120 if (destination.closed) {
125 const index = this.index++;
126 if (this.active < this.concurrent) {
127 destination.next!(value);
129 const { project } = this;
130 const result = project(value, index);
131 if (!this.scheduler) {
132 this.subscribeToProjection(result, value, index);
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));
139 destination.error!(e);
142 this.buffer!.push(value);
146 private subscribeToProjection(result: any, value: T, index: number): void {
148 const destination = this.destination as Subscription;
149 destination.add(innerSubscribe(result, new SimpleInnerSubscriber(this)));
152 protected _complete(): void {
153 this.hasCompleted = true;
154 if (this.hasCompleted && this.active === 0) {
155 this.destination.complete!();
160 notifyNext(innerValue: R): void {
161 this._next(innerValue);
164 notifyComplete(): void {
165 const buffer = this.buffer;
167 if (buffer && buffer.length > 0) {
168 this._next(buffer.shift());
170 if (this.hasCompleted && this.active === 0) {
171 this.destination.complete!();