import { Observable, ObservableInput } from '../Observable';
import { ArrayObservable } from '../observable/ArrayObservable';
import { isArray } from '../util/isArray';
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
const none = {};
/**
* 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` combines the values from this Observable with values from
* Observables passed as arguments. This is done by subscribing to each
* Observable, in order, and collecting an array of each of the most recent
* values any time any of the input Observables emits, then either taking that
* array and passing it as arguments to an optional `project` function and
* emitting the return value of that, or just emitting the array of recent
* values directly if there is no `project` function.
*
* @example
Dynamically calculate the Body-Mass Index from an Observable of weight and one for height
* var weight = Rx.Observable.of(70, 72, 76, 79, 75);
* var height = Rx.Observable.of(1.76, 1.77, 1.78);
* var bmi = weight.combineLatest(height, (w, h) => w / (h * h));
* bmi.subscribe(x => console.log('BMI is ' + x));
*
* @see {@link combineAll}
* @see {@link merge}
* @see {@link withLatestFrom}
*
* @param {Observable} other An input Observable to combine with the source
* Observable. More than one input Observables may be given as argument.
* @param {function} [project] An optional function to project the values from
* the combined latest values into a new value on the output 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.
* @method combineLatest
* @owner Observable
*/
export function combineLatest(...observables: Array |
Array> |
((...values: Array) => R)>): Observable {
let project: (...values: Array) => R = null;
if (typeof observables[observables.length - 1] === 'function') {
project = <(...values: Array) => R>observables.pop();
}
// if the first and only other argument besides the resultSelector is an array
// assume it's been called with `combineLatest([obs1, obs2, obs3], project)`
if (observables.length === 1 && isArray(observables[0])) {
observables = observables[0];
}
observables.unshift(this);
return new ArrayObservable(observables).lift(new CombineLatestOperator(project));
}
/* tslint:disable:max-line-length */
export interface CombineLatestSignature {
(project: (v1: T) => R): Observable;
(v2: ObservableInput, project: (v1: T, v2: T2) => R): Observable;
(v2: ObservableInput, v3: ObservableInput, project: (v1: T, v2: T2, v3: T3) => R): Observable;
(v2: ObservableInput, v3: ObservableInput, v4: ObservableInput, project: (v1: T, v2: T2, v3: T3, v4: T4) => R): Observable;
(v2: ObservableInput, v3: ObservableInput, v4: ObservableInput, v5: ObservableInput, project: (v1: T, v2: T2, v3: T3, v4: T4, v5: T5) => R): Observable;
(v2: ObservableInput, v3: ObservableInput, v4: ObservableInput, v5: ObservableInput, v6: ObservableInput, project: (v1: T, v2: T2, v3: T3, v4: T4, v5: T5, v6: T6) => R): Observable;
(v2: ObservableInput): Observable<[T, T2]>;
(v2: ObservableInput, v3: ObservableInput): Observable<[T, T2, T3]>;
(v2: ObservableInput, v3: ObservableInput, v4: ObservableInput): Observable<[T, T2, T3, T4]>;
(v2: ObservableInput, v3: ObservableInput, v4: ObservableInput, v5: ObservableInput): Observable<[T, T2, T3, T4, T5]>;
(v2: ObservableInput, v3: ObservableInput, v4: ObservableInput, v5: ObservableInput, v6: ObservableInput): Observable<[T, T2, T3, T4, T5, T6]>;
(...observables: Array | ((...values: Array) => R)>): Observable;
(array: ObservableInput[]): Observable>;
(array: ObservableInput[], project: (v1: T, ...values: Array) => R): Observable;
}
/* tslint:enable:max-line-length */
export class CombineLatestOperator implements Operator {
constructor(private project?: (...values: Array) => R) {
}
call(subscriber: Subscriber, source: any): any {
return source._subscribe(new CombineLatestSubscriber(subscriber, this.project));
}
}
/**
* 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 project?: (...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, observable, i));
}
}
}
notifyComplete(unused: Subscriber): void {
if ((this.active -= 1) === 0) {
this.destination.complete();
}
}
notifyNext(outerValue: T, innerValue: R,
outerIndex: number, innerIndex: number,
innerSub: InnerSubscriber): 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.project) {
this._tryProject(values);
} else {
this.destination.next(values.slice());
}
}
}
private _tryProject(values: any[]) {
let result: any;
try {
result = this.project.apply(this, values);
} catch (err) {
this.destination.error(err);
return;
}
this.destination.next(result);
}
}