import { Observable, SubscribableOrPromise } from '../Observable'; import { Subscriber } from '../Subscriber'; import { Subscription } from '../Subscription'; import { EmptyObservable } from './EmptyObservable'; import { isArray } from '../util/isArray'; import { subscribeToResult } from '../util/subscribeToResult'; import { OuterSubscriber } from '../OuterSubscriber'; import { InnerSubscriber } from '../InnerSubscriber'; /** * We need this JSDoc comment for affecting ESDoc. * @extends {Ignored} * @hide true */ export class ForkJoinObservable extends Observable { constructor(private sources: Array>, private resultSelector?: (...values: Array) => T) { super(); } /* tslint:disable:max-line-length */ static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise): Observable<[T, T2]>; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise): Observable<[T, T2, T3]>; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, v4: SubscribableOrPromise): Observable<[T, T2, T3, T4]>; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, v4: SubscribableOrPromise, v5: SubscribableOrPromise): Observable<[T, T2, T3, T4, T5]>; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, v4: SubscribableOrPromise, v5: SubscribableOrPromise, v6: SubscribableOrPromise): Observable<[T, T2, T3, T4, T5, T6]>; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, project: (v1: T, v2: T2) => R): Observable; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, project: (v1: T, v2: T2, v3: T3) => R): Observable; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, v4: SubscribableOrPromise, project: (v1: T, v2: T2, v3: T3, v4: T4) => R): Observable; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, v4: SubscribableOrPromise, v5: SubscribableOrPromise, project: (v1: T, v2: T2, v3: T3, v4: T4, v5: T5) => R): Observable; static create(v1: SubscribableOrPromise, v2: SubscribableOrPromise, v3: SubscribableOrPromise, v4: SubscribableOrPromise, v5: SubscribableOrPromise, v6: SubscribableOrPromise, project: (v1: T, v2: T2, v3: T3, v4: T4, v5: T5, v6: T6) => R): Observable; static create(sources: SubscribableOrPromise[]): Observable; static create(sources: SubscribableOrPromise[]): Observable; static create(sources: SubscribableOrPromise[], project: (...values: Array) => R): Observable; static create(sources: SubscribableOrPromise[], project: (...values: Array) => R): Observable; static create(...sources: SubscribableOrPromise[]): Observable; static create(...sources: SubscribableOrPromise[]): Observable; /* tslint:enable:max-line-length */ /** * @param sources * @return {any} * @static true * @name forkJoin * @owner Observable */ static create(...sources: Array | Array> | ((...values: Array) => any)>): Observable { if (sources === null || arguments.length === 0) { return new EmptyObservable(); } let resultSelector: (...values: Array) => any = null; if (typeof sources[sources.length - 1] === 'function') { resultSelector = <(...values: Array) => any>sources.pop(); } // if the first and only other argument besides the resultSelector is an array // assume it's been called with `forkJoin([obs1, obs2, obs3], resultSelector)` if (sources.length === 1 && isArray(sources[0])) { sources = >>sources[0]; } if (sources.length === 0) { return new EmptyObservable(); } return new ForkJoinObservable(>>sources, resultSelector); } protected _subscribe(subscriber: Subscriber): Subscription { return new ForkJoinSubscriber(subscriber, this.sources, this.resultSelector); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class ForkJoinSubscriber extends OuterSubscriber { private completed = 0; private total: number; private values: any[]; private haveValues = 0; constructor(destination: Subscriber, private sources: Array>, private resultSelector?: (...values: Array) => T) { super(destination); const len = sources.length; this.total = len; this.values = new Array(len); for (let i = 0; i < len; i++) { const source = sources[i]; const innerSubscription = subscribeToResult(this, source, null, i); if (innerSubscription) { ( innerSubscription).outerIndex = i; this.add(innerSubscription); } } } notifyNext(outerValue: any, innerValue: T, outerIndex: number, innerIndex: number, innerSub: InnerSubscriber): void { this.values[outerIndex] = innerValue; if (!(innerSub)._hasValue) { (innerSub)._hasValue = true; this.haveValues++; } } notifyComplete(innerSub: InnerSubscriber): void { const destination = this.destination; const { haveValues, resultSelector, values } = this; const len = values.length; if (!(innerSub)._hasValue) { destination.complete(); return; } this.completed++; if (this.completed !== len) { return; } if (haveValues === len) { const value = resultSelector ? resultSelector.apply(this, values) : values; destination.next(value); } destination.complete(); } }