/
combineLatest.ts
301 lines (289 loc) · 13.2 KB
/
combineLatest.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
import { Observable } from '../Observable';
import { ObservableInput, SchedulerLike, ObservedValueOf, ObservableInputTuple } from '../types';
import { argsArgArrayOrObject } from '../util/argsArgArrayOrObject';
import { Subscriber } from '../Subscriber';
import { from } from './from';
import { identity } from '../util/identity';
import { Subscription } from '../Subscription';
import { mapOneOrManyArgs } from '../util/mapOneOrManyArgs';
import { popResultSelector, popScheduler } from '../util/args';
import { createObject } from '../util/createObject';
import { OperatorSubscriber } from '../operators/OperatorSubscriber';
import { AnyCatcher } from '../AnyCatcher';
import { executeSchedule } from '../util/executeSchedule';
// combineLatest(any)
// We put this first because we need to catch cases where the user has supplied
// _exactly `any`_ as the argument. Since `any` literally matches _anything_,
// we don't want it to randomly hit one of the other type signatures below,
// as we have no idea at build-time what type we should be returning when given an any.
/**
* You have passed `any` here, we can't figure out if it is
* an array or an object, so you're getting `unknown`. Use better types.
* @param arg Something typed as `any`
*/
export function combineLatest<T extends AnyCatcher>(arg: T): Observable<unknown>;
// combineLatest([a, b, c])
export function combineLatest(sources: []): Observable<never>;
export function combineLatest<A extends readonly unknown[]>(sources: readonly [...ObservableInputTuple<A>]): Observable<A>;
/** @deprecated The `scheduler` parameter will be removed in v8. Use `scheduled` and `combineLatestAll`. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function combineLatest<A extends readonly unknown[], R>(
sources: readonly [...ObservableInputTuple<A>],
resultSelector: (...values: A) => R,
scheduler: SchedulerLike
): Observable<R>;
export function combineLatest<A extends readonly unknown[], R>(
sources: readonly [...ObservableInputTuple<A>],
resultSelector: (...values: A) => R
): Observable<R>;
/** @deprecated The `scheduler` parameter will be removed in v8. Use `scheduled` and `combineLatestAll`. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function combineLatest<A extends readonly unknown[]>(
sources: readonly [...ObservableInputTuple<A>],
scheduler: SchedulerLike
): Observable<A>;
// combineLatest(a, b, c)
/** @deprecated Pass an array of sources instead. The rest-parameters signature will be removed in v8. Details: https://rxjs.dev/deprecations/array-argument */
export function combineLatest<A extends readonly unknown[]>(...sources: [...ObservableInputTuple<A>]): Observable<A>;
/** @deprecated The `scheduler` parameter will be removed in v8. Use `scheduled` and `combineLatestAll`. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function combineLatest<A extends readonly unknown[], R>(
...sourcesAndResultSelectorAndScheduler: [...ObservableInputTuple<A>, (...values: A) => R, SchedulerLike]
): Observable<R>;
/** @deprecated Pass an array of sources instead. The rest-parameters signature will be removed in v8. Details: https://rxjs.dev/deprecations/array-argument */
export function combineLatest<A extends readonly unknown[], R>(
...sourcesAndResultSelector: [...ObservableInputTuple<A>, (...values: A) => R]
): Observable<R>;
/** @deprecated The `scheduler` parameter will be removed in v8. Use `scheduled` and `combineLatestAll`. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function combineLatest<A extends readonly unknown[]>(
...sourcesAndScheduler: [...ObservableInputTuple<A>, SchedulerLike]
): Observable<A>;
// combineLatest({a, b, c})
export function combineLatest(sourcesObject: { [K in any]: never }): Observable<never>;
export function combineLatest<T extends Record<string, ObservableInput<any>>>(
sourcesObject: T
): Observable<{ [K in keyof T]: ObservedValueOf<T[K]> }>;
/**
* Combines multiple Observables to create an Observable whose values are
* calculated from the latest values of each of its input Observables.
*
* <span class="informal">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.</span>
*
* ![](combineLatest.png)
*
* `combineLatest` combines the values from all the Observables passed in the
* observables array. 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 this operator,
* the returned Observable will always emit an array of `n` values, in an order
* corresponding to the order of the passed Observables (the value from the first Observable
* will be at index 0 of the array and so on).
*
* Static version of `combineLatest` accepts an array of Observables. Note that an array of
* Observables is a good choice, if you don't know beforehand how many Observables
* you will combine. Passing an empty array will result in an Observable that
* completes immediately.
*
* To ensure the output array always has 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 now be impossible to include a value from the
* completed Observable in the 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, the resulting Observable will complete when all combined
* streams complete. So even if some Observable completes, the result of
* `combineLatest` will still emit values when other Observables do. In case
* of a 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.
*
* ## 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 a dictionary of Observables
* ```ts
* import { combineLatest, of } from 'rxjs';
* import { delay, startWith } from 'rxjs/operators';
*
* const observables = {
* a: of(1).pipe(delay(1000), startWith(0)),
* b: of(5).pipe(delay(5000), startWith(0)),
* c: of(10).pipe(delay(10000), startWith(0))
* };
* const combined = combineLatest(observables);
* combined.subscribe(value => console.log(value));
* // Logs
* // {a: 0, b: 0, c: 0} immediately
* // {a: 1, b: 0, c: 0} after 1s
* // {a: 1, b: 5, c: 0} after 5s
* // {a: 1, b: 5, c: 10} after 10s
* ```
* ### Combine an array of Observables
* ```ts
* import { combineLatest, of } from 'rxjs';
* import { delay, startWith } 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 map operator 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 combineLatestAll}
* @see {@link merge}
* @see {@link withLatestFrom}
*
* @param {ObservableInput} [observables] An array of input Observables to combine with each other.
* An array of Observables must 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<O extends ObservableInput<any>, R>(...args: any[]): Observable<R> | Observable<ObservedValueOf<O>[]> {
const scheduler = popScheduler(args);
const resultSelector = popResultSelector(args);
const { args: observables, keys } = argsArgArrayOrObject(args);
if (observables.length === 0) {
// If no observables are passed, or someone has passed an ampty array
// of observables, or even an empty object POJO, we need to just
// complete (EMPTY), but we have to honor the scheduler provided if any.
return from([], scheduler as any);
}
const result = new Observable<ObservedValueOf<O>[]>(
combineLatestInit(
observables as ObservableInput<ObservedValueOf<O>>[],
scheduler,
keys
? // A handler for scrubbing the array of args into a dictionary.
(values) => createObject(keys, values)
: // A passthrough to just return the array
identity
)
);
return resultSelector ? (result.pipe(mapOneOrManyArgs(resultSelector)) as Observable<R>) : result;
}
export function combineLatestInit(
observables: ObservableInput<any>[],
scheduler?: SchedulerLike,
valueTransform: (values: any[]) => any = identity
) {
return (subscriber: Subscriber<any>) => {
// The outer subscription. We're capturing this in a function
// because we may have to schedule it.
maybeSchedule(
scheduler,
() => {
const { length } = observables;
// A store for the values each observable has emitted so far. We match observable to value on index.
const values = new Array(length);
// The number of currently active subscriptions, as they complete, we decrement this number to see if
// we are all done combining values, so we can complete the result.
let active = length;
// The number of inner sources that still haven't emitted the first value
// We need to track this because all sources need to emit one value in order
// to start emitting values.
let remainingFirstValues = length;
// The loop to kick off subscription. We're keying everything on index `i` to relate the observables passed
// in to the slot in the output array or the key in the array of keys in the output dictionary.
for (let i = 0; i < length; i++) {
maybeSchedule(
scheduler,
() => {
const source = from(observables[i], scheduler as any);
let hasFirstValue = false;
source.subscribe(
new OperatorSubscriber(
subscriber,
(value) => {
// When we get a value, record it in our set of values.
values[i] = value;
if (!hasFirstValue) {
// If this is our first value, record that.
hasFirstValue = true;
remainingFirstValues--;
}
if (!remainingFirstValues) {
// We're not waiting for any more
// first values, so we can emit!
subscriber.next(valueTransform(values.slice()));
}
},
() => {
if (!--active) {
// We only complete the result if we have no more active
// inner observables.
subscriber.complete();
}
}
)
);
},
subscriber
);
}
},
subscriber
);
};
}
/**
* A small utility to handle the couple of locations where we want to schedule if a scheduler was provided,
* but we don't if there was no scheduler.
*/
function maybeSchedule(scheduler: SchedulerLike | undefined, execute: () => void, subscription: Subscription) {
if (scheduler) {
executeSchedule(subscription, scheduler, execute);
} else {
execute();
}
}