-
Notifications
You must be signed in to change notification settings - Fork 3k
/
concat.ts
123 lines (119 loc) · 6.51 KB
/
concat.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
import {Observable, ObservableInput} from '../Observable';
import {Scheduler} from '../Scheduler';
import {isScheduler} from '../util/isScheduler';
import {ArrayObservable} from '../observable/ArrayObservable';
import {MergeAllOperator} from './mergeAll';
/**
* Creates an output Observable which sequentially emits all values from every
* given input Observable after the current Observable.
*
* <span class="informal">Concatenates multiple Observables together by
* sequentially emitting their values, one Observable after the other.</span>
*
* <img src="./img/concat.png" width="100%">
*
* Joins this Observable with multiple other Observables by subscribing to them
* one at a time, starting with the source, and merging their results into the
* output Observable. Will wait for each Observable to complete before moving
* on to the next.
*
* @example <caption>Concatenate a timer counting from 0 to 3 with a synchronous sequence from 1 to 10</caption>
* var timer = Rx.Observable.interval(1000).take(4);
* var sequence = Rx.Observable.range(1, 10);
* var result = timer.concat(sequence);
* result.subscribe(x => console.log(x));
*
* @example <caption>Concatenate 3 Observables</caption>
* var timer1 = Rx.Observable.interval(1000).take(10);
* var timer2 = Rx.Observable.interval(2000).take(6);
* var timer3 = Rx.Observable.interval(500).take(10);
* var result = timer1.concat(timer2, timer3);
* result.subscribe(x => console.log(x));
*
* @see {@link concatAll}
* @see {@link concatMap}
* @see {@link concatMapTo}
*
* @param {Observable} other An input Observable to concatenate after the source
* Observable. More than one input Observables may be given as argument.
* @param {Scheduler} [scheduler=null] An optional Scheduler to schedule each
* Observable subscription on.
* @return {Observable} All values of each passed Observable merged into a
* single Observable, in order, in serial fashion.
* @method concat
* @owner Observable
*/
export function concat<T, R>(...observables: Array<ObservableInput<any> | Scheduler>): Observable<R> {
return concatStatic<T, R>(this, ...observables);
}
/* tslint:disable:max-line-length */
export interface ConcatSignature<T> {
(scheduler?: Scheduler): Observable<T>;
<T2>(v2: ObservableInput<T2>, scheduler?: Scheduler): Observable<T | T2>;
<T2, T3>(v2: ObservableInput<T2>, v3: ObservableInput<T3>, scheduler?: Scheduler): Observable<T | T2 | T3>;
<T2, T3, T4>(v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, scheduler?: Scheduler): Observable<T | T2 | T3 | T4>;
<T2, T3, T4, T5>(v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, scheduler?: Scheduler): Observable<T | T2 | T3 | T4 | T5>;
<T2, T3, T4, T5, T6>(v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>, scheduler?: Scheduler): Observable<T | T2 | T3 | T4 | T5 | T6>;
(...observables: Array<ObservableInput<T> | Scheduler>): Observable<T>;
<R>(...observables: Array<ObservableInput<any> | Scheduler>): Observable<R>;
}
/* tslint:enable:max-line-length */
/* tslint:disable:max-line-length */
export function concatStatic<T>(v1: ObservableInput<T>, scheduler?: Scheduler): Observable<T>;
export function concatStatic<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>, scheduler?: Scheduler): Observable<T | T2>;
export function concatStatic<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, scheduler?: Scheduler): Observable<T | T2 | T3>;
export function concatStatic<T, T2, T3, T4>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, scheduler?: Scheduler): Observable<T | T2 | T3 | T4>;
export function concatStatic<T, T2, T3, T4, T5>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, scheduler?: Scheduler): Observable<T | T2 | T3 | T4 | T5>;
export function concatStatic<T, T2, T3, T4, T5, T6>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>, scheduler?: Scheduler): Observable<T | T2 | T3 | T4 | T5 | T6>;
export function concatStatic<T>(...observables: (ObservableInput<T> | Scheduler)[]): Observable<T>;
export function concatStatic<T, R>(...observables: (ObservableInput<any> | Scheduler)[]): Observable<R>;
/* tslint:enable:max-line-length */
/**
* Creates an output Observable which sequentially emits all values from every
* given input Observable after the current Observable.
*
* <span class="informal">Concatenates multiple Observables together by
* sequentially emitting their values, one Observable after the other.</span>
*
* <img src="./img/concat.png" width="100%">
*
* Joins multiple Observables together by subscribing to them one at a time and
* merging their results into the output Observable. Will wait for each
* Observable to complete before moving on to the next.
*
* @example <caption>Concatenate a timer counting from 0 to 3 with a synchronous sequence from 1 to 10</caption>
* var timer = Rx.Observable.interval(1000).take(4);
* var sequence = Rx.Observable.range(1, 10);
* var result = Rx.Observable.concat(timer, sequence);
* result.subscribe(x => console.log(x));
*
* @example <caption>Concatenate 3 Observables</caption>
* var timer1 = Rx.Observable.interval(1000).take(10);
* var timer2 = Rx.Observable.interval(2000).take(6);
* var timer3 = Rx.Observable.interval(500).take(10);
* var result = Rx.Observable.concat(timer1, timer2, timer3);
* result.subscribe(x => console.log(x));
*
* @see {@link concatAll}
* @see {@link concatMap}
* @see {@link concatMapTo}
*
* @param {Observable} input1 An input Observable to concatenate with others.
* @param {Observable} input2 An input Observable to concatenate with others.
* More than one input Observables may be given as argument.
* @param {Scheduler} [scheduler=null] An optional Scheduler to schedule each
* Observable subscription on.
* @return {Observable} All values of each passed Observable merged into a
* single Observable, in order, in serial fashion.
* @static true
* @name concat
* @owner Observable
*/
export function concatStatic<T, R>(...observables: Array<ObservableInput<any> | Scheduler>): Observable<R> {
let scheduler: Scheduler = null;
let args = <any[]>observables;
if (isScheduler(args[observables.length - 1])) {
scheduler = args.pop();
}
return new ArrayObservable(observables, scheduler).lift(new MergeAllOperator<R>(1));
}