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(); } /** * @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(); } }