import { Observable } from '../Observable'; import { ObservableInput, SchedulerLike, ObservedValueOf } from '../types'; import { isScheduler } from '../util/isScheduler'; import { isArray } from '../util/isArray'; import { Subscriber } from '../Subscriber'; import { OuterSubscriber } from '../OuterSubscriber'; import { Operator } from '../Operator'; import { InnerSubscriber } from '../InnerSubscriber'; import { subscribeToResult } from '../util/subscribeToResult'; import { fromArray } from './fromArray'; const NONE = {}; /* tslint:disable:max-line-length */ // If called with a single array, it "auto-spreads" the array, with result selector /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, R>(sources: [O1], resultSelector: (v1: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, R>(sources: [O1, O2], resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, R>(sources: [O1, O2, O3], resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, R>(sources: [O1, O2, O3, O4], resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf, v4: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, R>(sources: [O1, O2, O3, O4, O5], resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf, v4: ObservedValueOf, v5: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, O6 extends ObservableInput, R>(sources: [O1, O2, O3, O4, O5, O6], resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf, v4: ObservedValueOf, v5: ObservedValueOf, v6: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, R>(sources: O[], resultSelector: (...args: ObservedValueOf[]) => R, scheduler?: SchedulerLike): Observable; // standard call, but with a result selector /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, R>(v1: O1, resultSelector: (v1: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, R>(v1: O1, v2: O2, resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, R>(v1: O1, v2: O2, v3: O3, resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, R>(v1: O1, v2: O2, v3: O3, v4: O4, resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf, v4: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, R>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf, v4: ObservedValueOf, v5: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, O6 extends ObservableInput, R>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, v6: O6, resultSelector: (v1: ObservedValueOf, v2: ObservedValueOf, v3: ObservedValueOf, v4: ObservedValueOf, v5: ObservedValueOf, v6: ObservedValueOf) => R, scheduler?: SchedulerLike): Observable; // With a scheduler (deprecated) /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest>(sources: [O1], scheduler: SchedulerLike): Observable<[ObservedValueOf]>; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest, O2 extends ObservableInput>(sources: [O1, O2], scheduler: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf]>; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput>(sources: [O1, O2, O3], scheduler: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput>(sources: [O1, O2, O3, O4], scheduler: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput>(sources: [O1, O2, O3, O4, O5], scheduler: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, O6 extends ObservableInput>(sources: [O1, O2, O3, O4, O5, O6], scheduler: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest>(sources: O[], scheduler: SchedulerLike): Observable[]>; // Best case export function combineLatest>(sources: [O1]): Observable<[ObservedValueOf]>; export function combineLatest, O2 extends ObservableInput>(sources: [O1, O2]): Observable<[ObservedValueOf, ObservedValueOf]>; export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput>(sources: [O1, O2, O3]): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf]>; export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput>(sources: [O1, O2, O3, O4]): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput>(sources: [O1, O2, O3, O4, O5]): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, O6 extends ObservableInput>(sources: [O1, O2, O3, O4, O5, O6]): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; export function combineLatest>(sources: O[]): Observable[]>; // Standard calls /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest>(v1: O1, scheduler?: SchedulerLike): Observable<[ObservedValueOf]>; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest, O2 extends ObservableInput>(v1: O1, v2: O2, scheduler?: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf]>; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput>(v1: O1, v2: O2, v3: O3, scheduler?: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput>(v1: O1, v2: O2, v3: O3, v4: O4, scheduler?: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, scheduler?: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest, O2 extends ObservableInput, O3 extends ObservableInput, O4 extends ObservableInput, O5 extends ObservableInput, O6 extends ObservableInput>(v1: O1, v2: O2, v3: O3, v4: O4, v5: O5, v6: O6, scheduler?: SchedulerLike): Observable<[ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf, ObservedValueOf]>; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest>(...observables: O[]): Observable; /** @deprecated Pass arguments in a single array instead `combineLatest([a, b, c])` */ export function combineLatest, R>(...observables: Array | ((...values: Array) => R)>): Observable; /** @deprecated resultSelector no longer supported, pipe to map instead */ export function combineLatest, R>(array: O[], resultSelector: (...values: ObservedValueOf[]) => R, scheduler?: SchedulerLike): Observable; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest>(...observables: Array): Observable; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest, R>(...observables: Array[]) => R) | SchedulerLike>): Observable; /** @deprecated Passing a scheduler here is deprecated, use {@link subscribeOn} and/or {@link observeOn} instead */ export function combineLatest(...observables: Array | ((...values: Array) => R) | SchedulerLike>): Observable; /* tslint:enable:max-line-length */ /** * Combines multiple Observables to create an Observable whose values are * calculated from the latest values of each of its input Observables. * * Whenever any input Observable emits a value, it * computes a formula using the latest values from all the inputs, then emits * the output of that formula. * * ![](combineLatest.png) * * `combineLatest` combines the values from all the Observables passed as * arguments. This is done by subscribing to each Observable in order and, * whenever any Observable emits, collecting an array of the most recent * values from each Observable. So if you pass `n` Observables to operator, * returned Observable will always emit an array of `n` values, in order * corresponding to order of passed Observables (value from the first Observable * on the first place and so on). * * Static version of `combineLatest` accepts either an array of Observables * or each Observable can be put directly as an argument. Note that array of * Observables is good choice, if you don't know beforehand how many Observables * you will combine. Passing empty array will result in Observable that * completes immediately. * * To ensure output array has always the same length, `combineLatest` will * actually wait for all input Observables to emit at least once, * before it starts emitting results. This means if some Observable emits * values before other Observables started emitting, all these values but the last * will be lost. On the other hand, if some Observable does not emit a value but * completes, resulting Observable will complete at the same moment without * emitting anything, since it will be now impossible to include value from * completed Observable in resulting array. Also, if some input Observable does * not emit any value and never completes, `combineLatest` will also never emit * and never complete, since, again, it will wait for all streams to emit some * value. * * If at least one Observable was passed to `combineLatest` and all passed Observables * emitted something, resulting Observable will complete when all combined * streams complete. So even if some Observable completes, result of * `combineLatest` will still emit values when other Observables do. In case * of completed Observable, its value from now on will always be the last * emitted value. On the other hand, if any Observable errors, `combineLatest` * will error immediately as well, and all other Observables will be unsubscribed. * * `combineLatest` accepts as optional parameter `project` function, which takes * as arguments all values that would normally be emitted by resulting Observable. * `project` can return any kind of value, which will be then emitted by Observable * instead of default array. Note that `project` does not take as argument that array * of values, but values themselves. That means default `project` can be imagined * as function that takes all its arguments and puts them into an array. * * ## Examples * ### Combine two timer Observables * ```ts * import { combineLatest, timer } from 'rxjs'; * * const firstTimer = timer(0, 1000); // emit 0, 1, 2... after every second, starting from now * const secondTimer = timer(500, 1000); // emit 0, 1, 2... after every second, starting 0,5s from now * const combinedTimers = combineLatest(firstTimer, secondTimer); * combinedTimers.subscribe(value => console.log(value)); * // Logs * // [0, 0] after 0.5s * // [1, 0] after 1s * // [1, 1] after 1.5s * // [2, 1] after 2s * ``` * * ### Combine an array of Observables * ```ts * import { combineLatest, of } from 'rxjs'; * import { delay, starWith } from 'rxjs/operators'; * * const observables = [1, 5, 10].map( * n => of(n).pipe( * delay(n * 1000), // emit 0 and then emit n after n seconds * startWith(0), * ) * ); * const combined = combineLatest(observables); * combined.subscribe(value => console.log(value)); * // Logs * // [0, 0, 0] immediately * // [1, 0, 0] after 1s * // [1, 5, 0] after 5s * // [1, 5, 10] after 10s * ``` * * * ### Use project function to dynamically calculate the Body-Mass Index * ```ts * import { combineLatest, of } from 'rxjs'; * import { map } from 'rxjs/operators'; * * const weight = of(70, 72, 76, 79, 75); * const height = of(1.76, 1.77, 1.78); * const bmi = combineLatest(weight, height).pipe( * map(([w, h]) => w / (h * h)), * ); * bmi.subscribe(x => console.log('BMI is ' + x)); * * // With output to console: * // BMI is 24.212293388429753 * // BMI is 23.93948099205209 * // BMI is 23.671253629592222 * ``` * * @see {@link combineAll} * @see {@link merge} * @see {@link withLatestFrom} * * @param {ObservableInput} observable1 An input Observable to combine with other Observables. * @param {ObservableInput} observable2 An input Observable to combine with other Observables. * More than one input Observables may be given as arguments * or an array of Observables may be given as the first argument. * @param {function} [project] An optional function to project the values from * the combined latest values into a new value on the output Observable. * @param {SchedulerLike} [scheduler=null] The {@link SchedulerLike} to use for subscribing to * each input Observable. * @return {Observable} An Observable of projected values from the most recent * values from each input Observable, or an array of the most recent values from * each input Observable. */ export function combineLatest, R>( ...observables: (O | ((...values: ObservedValueOf[]) => R) | SchedulerLike)[] ): Observable { let resultSelector: ((...values: Array) => R) | undefined = undefined; let scheduler: SchedulerLike|undefined = undefined; if (isScheduler(observables[observables.length - 1])) { scheduler = observables.pop() as SchedulerLike; } if (typeof observables[observables.length - 1] === 'function') { resultSelector = observables.pop() as (...values: Array) => R; } // if the first and only other argument besides the resultSelector is an array // assume it's been called with `combineLatest([obs1, obs2, obs3], resultSelector)` if (observables.length === 1 && isArray(observables[0])) { observables = observables[0] as any; } return fromArray(observables, scheduler).lift(new CombineLatestOperator(resultSelector)); } export class CombineLatestOperator implements Operator { constructor(private resultSelector?: (...values: Array) => R) { } call(subscriber: Subscriber, source: any): any { return source.subscribe(new CombineLatestSubscriber(subscriber, this.resultSelector)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ export class CombineLatestSubscriber extends OuterSubscriber { private active: number = 0; private values: any[] = []; private observables: any[] = []; private toRespond?: number; constructor(destination: Subscriber, private resultSelector?: (...values: Array) => R) { super(destination); } protected _next(observable: any) { this.values.push(NONE); this.observables.push(observable); } protected _complete() { const observables = this.observables; const len = observables.length; if (len === 0) { this.destination.complete!(); } else { this.active = len; this.toRespond = len; for (let i = 0; i < len; i++) { const observable = observables[i]; this.add(subscribeToResult(this, observable, undefined, i)); } } } notifyComplete(unused: Subscriber): void { if ((this.active -= 1) === 0) { this.destination.complete!(); } } notifyNext(_outerValue: T, innerValue: R, outerIndex: number): void { const values = this.values; const oldVal = values[outerIndex]; const toRespond = !this.toRespond ? 0 : oldVal === NONE ? --this.toRespond : this.toRespond; values[outerIndex] = innerValue; if (toRespond === 0) { if (this.resultSelector) { this._tryResultSelector(values); } else { this.destination.next!(values.slice()); } } } private _tryResultSelector(values: any[]) { let result: any; try { result = this.resultSelector!.apply(this, values); } catch (err) { this.destination.error!(err); return; } this.destination.next!(result); } }