--- /dev/null
+import { Observable } from '../Observable';
+import { Operator } from '../Operator';
+import { Subscriber } from '../Subscriber';
+import { Subscription } from '../Subscription';
+import { OuterSubscriber } from '../OuterSubscriber';
+import { InnerSubscriber } from '../InnerSubscriber';
+import { subscribeToResult } from '../util/subscribeToResult';
+import { MonoTypeOperatorFunction, OperatorFunction, ObservableInput, SchedulerLike } from '../types';
+
+/* tslint:disable:max-line-length */
+export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>, concurrent?: number, scheduler?: SchedulerLike): OperatorFunction<T, R>;
+export function expand<T>(project: (value: T, index: number) => ObservableInput<T>, concurrent?: number, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>;
+/* tslint:enable:max-line-length */
+
+/**
+ * Recursively projects each source value to an Observable which is merged in
+ * the output Observable.
+ *
+ * <span class="informal">It's similar to {@link mergeMap}, but applies the
+ * projection function to every source value as well as every output value.
+ * It's recursive.</span>
+ *
+ * 
+ *
+ * Returns an Observable that emits items based on applying a function that you
+ * supply to each item emitted by the source Observable, where that function
+ * returns an Observable, and then merging those resulting Observables and
+ * emitting the results of this merger. *Expand* will re-emit on the output
+ * Observable every source value. Then, each output value is given to the
+ * `project` function which returns an inner Observable to be merged on the
+ * output Observable. Those output values resulting from the projection are also
+ * given to the `project` function to produce new output values. This is how
+ * *expand* behaves recursively.
+ *
+ * ## Example
+ * Start emitting the powers of two on every click, at most 10 of them
+ * ```ts
+ * import { fromEvent, of } from 'rxjs';
+ * import { expand, mapTo, delay, take } from 'rxjs/operators';
+ *
+ * const clicks = fromEvent(document, 'click');
+ * const powersOfTwo = clicks.pipe(
+ * mapTo(1),
+ * expand(x => of(2 * x).pipe(delay(1000))),
+ * take(10),
+ * );
+ * powersOfTwo.subscribe(x => console.log(x));
+ * ```
+ *
+ * @see {@link mergeMap}
+ * @see {@link mergeScan}
+ *
+ * @param {function(value: T, index: number) => Observable} project A function
+ * that, when applied to an item emitted by the source or the output Observable,
+ * returns an Observable.
+ * @param {number} [concurrent=Number.POSITIVE_INFINITY] Maximum number of input
+ * Observables being subscribed to concurrently.
+ * @param {SchedulerLike} [scheduler=null] The {@link SchedulerLike} to use for subscribing to
+ * each projected inner Observable.
+ * @return {Observable} An Observable that emits the source values and also
+ * result of applying the projection function to each value emitted on the
+ * output Observable and and merging the results of the Observables obtained
+ * from this transformation.
+ * @method expand
+ * @owner Observable
+ */
+export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>,
+ concurrent: number = Number.POSITIVE_INFINITY,
+ scheduler: SchedulerLike = undefined): OperatorFunction<T, R> {
+ concurrent = (concurrent || 0) < 1 ? Number.POSITIVE_INFINITY : concurrent;
+
+ return (source: Observable<T>) => source.lift(new ExpandOperator(project, concurrent, scheduler));
+}
+
+export class ExpandOperator<T, R> implements Operator<T, R> {
+ constructor(private project: (value: T, index: number) => ObservableInput<R>,
+ private concurrent: number,
+ private scheduler: SchedulerLike) {
+ }
+
+ call(subscriber: Subscriber<R>, source: any): any {
+ return source.subscribe(new ExpandSubscriber(subscriber, this.project, this.concurrent, this.scheduler));
+ }
+}
+
+interface DispatchArg<T, R> {
+ subscriber: ExpandSubscriber<T, R>;
+ result: ObservableInput<R>;
+ value: any;
+ index: number;
+}
+
+/**
+ * We need this JSDoc comment for affecting ESDoc.
+ * @ignore
+ * @extends {Ignored}
+ */
+export class ExpandSubscriber<T, R> extends OuterSubscriber<T, R> {
+ private index: number = 0;
+ private active: number = 0;
+ private hasCompleted: boolean = false;
+ private buffer: any[];
+
+ constructor(destination: Subscriber<R>,
+ private project: (value: T, index: number) => ObservableInput<R>,
+ private concurrent: number,
+ private scheduler: SchedulerLike) {
+ super(destination);
+ if (concurrent < Number.POSITIVE_INFINITY) {
+ this.buffer = [];
+ }
+ }
+
+ private static dispatch<T, R>(arg: DispatchArg<T, R>): void {
+ const {subscriber, result, value, index} = arg;
+ subscriber.subscribeToProjection(result, value, index);
+ }
+
+ protected _next(value: any): void {
+ const destination = this.destination;
+
+ if (destination.closed) {
+ this._complete();
+ return;
+ }
+
+ const index = this.index++;
+ if (this.active < this.concurrent) {
+ destination.next(value);
+ try {
+ const { project } = this;
+ const result = project(value, index);
+ if (!this.scheduler) {
+ this.subscribeToProjection(result, value, index);
+ } else {
+ const state: DispatchArg<T, R> = { subscriber: this, result, value, index };
+ const destination = this.destination as Subscription;
+ destination.add(this.scheduler.schedule<DispatchArg<T, R>>(ExpandSubscriber.dispatch, 0, state));
+ }
+ } catch (e) {
+ destination.error(e);
+ }
+ } else {
+ this.buffer.push(value);
+ }
+ }
+
+ private subscribeToProjection(result: any, value: T, index: number): void {
+ this.active++;
+ const destination = this.destination as Subscription;
+ destination.add(subscribeToResult<T, R>(this, result, value, index));
+ }
+
+ protected _complete(): void {
+ this.hasCompleted = true;
+ if (this.hasCompleted && this.active === 0) {
+ this.destination.complete();
+ }
+ this.unsubscribe();
+ }
+
+ notifyNext(outerValue: T, innerValue: R,
+ outerIndex: number, innerIndex: number,
+ innerSub: InnerSubscriber<T, R>): void {
+ this._next(innerValue);
+ }
+
+ notifyComplete(innerSub: Subscription): void {
+ const buffer = this.buffer;
+ const destination = this.destination as Subscription;
+ destination.remove(innerSub);
+ this.active--;
+ if (buffer && buffer.length > 0) {
+ this._next(buffer.shift());
+ }
+ if (this.hasCompleted && this.active === 0) {
+ this.destination.complete();
+ }
+ }
+}
projects / dotfiles / .git / blobdiff