|
1 | | -import {Observable} from '../Observable'; |
2 | | -import {Subscriber} from '../Subscriber'; |
3 | | -import {PromiseObservable} from './PromiseObservable'; |
4 | | -import {EmptyObservable} from './EmptyObservable'; |
5 | | -import {isPromise} from '../util/isPromise'; |
6 | | - |
7 | | -export class ForkJoinObservable<T> extends Observable<T> { |
8 | | - constructor(private sources: Array<Observable<any> | |
9 | | - Promise<any> | |
10 | | - ((...values: Array<any>) => any)>) { |
11 | | - super(); |
12 | | - } |
13 | | - |
14 | | - static create(...sources: Array<Observable<any> | |
15 | | - Promise<any> | |
16 | | - ((...values: Array<any>) => any)>) |
17 | | - : Observable<any> { |
18 | | - if (sources === null || sources.length === 0) { |
19 | | - return new EmptyObservable(); |
20 | | - } |
21 | | - return new ForkJoinObservable(sources); |
22 | | - } |
23 | | - |
24 | | - private getResultSelector(): (...values: Array<any>) => any { |
25 | | - const sources = this.sources; |
26 | | - |
27 | | - let resultSelector = sources[sources.length - 1]; |
28 | | - if (typeof resultSelector !== 'function') { |
29 | | - return null; |
30 | | - } |
31 | | - this.sources.pop(); |
32 | | - return <(...values: Array<any>) => any>resultSelector; |
33 | | - } |
34 | | - |
35 | | - _subscribe(subscriber: Subscriber<any>) { |
36 | | - let resultSelector = this.getResultSelector(); |
37 | | - const sources = this.sources; |
38 | | - const len = sources.length; |
39 | | - |
40 | | - const context = { completed: 0, total: len, values: emptyArray(len), selector: resultSelector }; |
41 | | - for (let i = 0; i < len; i++) { |
42 | | - let source = sources[i]; |
43 | | - if (isPromise(source)) { |
44 | | - source = new PromiseObservable(<Promise<any>>source); |
45 | | - } |
46 | | - (<Observable<any>>source).subscribe(new AllSubscriber(subscriber, i, context)); |
47 | | - } |
48 | | - } |
49 | | -} |
50 | | - |
51 | | -class AllSubscriber<T> extends Subscriber<T> { |
52 | | - private _value: any = null; |
53 | | - |
54 | | - constructor(destination: Subscriber<any>, |
55 | | - private index: number, |
56 | | - private context: { completed: number, |
57 | | - total: number, |
58 | | - values: any[], |
59 | | - selector: (...values: Array<any>) => any }) { |
60 | | - super(destination); |
61 | | - } |
62 | | - |
63 | | - _next(value: any): void { |
64 | | - this._value = value; |
65 | | - } |
66 | | - |
67 | | - _complete(): void { |
68 | | - const destination = this.destination; |
69 | | - |
70 | | - if (this._value == null) { |
71 | | - destination.complete(); |
72 | | - } |
73 | | - |
74 | | - const context = this.context; |
75 | | - context.completed++; |
76 | | - context.values[this.index] = this._value; |
77 | | - const values = context.values; |
78 | | - |
79 | | - if (context.completed !== values.length) { |
80 | | - return; |
81 | | - } |
82 | | - |
83 | | - if (values.every(hasValue)) { |
84 | | - let value = context.selector ? context.selector.apply(this, values) : |
85 | | - values; |
86 | | - destination.next(value); |
87 | | - } |
88 | | - |
89 | | - destination.complete(); |
90 | | - } |
91 | | -} |
92 | | - |
93 | | -function hasValue(x: any): boolean { |
94 | | - return x !== null; |
95 | | -} |
96 | | - |
97 | | -function emptyArray(len: number): any[] { |
98 | | - let arr = []; |
99 | | - for (let i = 0; i < len; i++) { |
100 | | - arr.push(null); |
101 | | - } |
102 | | - return arr; |
103 | | -} |
| 1 | +import {Observable} from '../Observable'; |
| 2 | +import {Subscriber} from '../Subscriber'; |
| 3 | +import {PromiseObservable} from './PromiseObservable'; |
| 4 | +import {EmptyObservable} from './EmptyObservable'; |
| 5 | +import {isPromise} from '../util/isPromise'; |
| 6 | + |
| 7 | +export class ForkJoinObservable<T> extends Observable<T> { |
| 8 | + constructor(private sources: Array<Observable<any> | |
| 9 | + Promise<any> | |
| 10 | + ((...values: Array<any>) => any)>) { |
| 11 | + super(); |
| 12 | + } |
| 13 | + |
| 14 | + static create(...sources: Array<Observable<any> | |
| 15 | + Promise<any> | |
| 16 | + ((...values: Array<any>) => any)>) |
| 17 | + : Observable<any> { |
| 18 | + if (sources === null || sources.length === 0) { |
| 19 | + return new EmptyObservable(); |
| 20 | + } |
| 21 | + return new ForkJoinObservable(sources); |
| 22 | + } |
| 23 | + |
| 24 | + private getResultSelector(): (...values: Array<any>) => any { |
| 25 | + const sources = this.sources; |
| 26 | + |
| 27 | + let resultSelector = sources[sources.length - 1]; |
| 28 | + if (typeof resultSelector !== 'function') { |
| 29 | + return null; |
| 30 | + } |
| 31 | + this.sources.pop(); |
| 32 | + return <(...values: Array<any>) => any>resultSelector; |
| 33 | + } |
| 34 | + |
| 35 | + _subscribe(subscriber: Subscriber<any>) { |
| 36 | + let resultSelector = this.getResultSelector(); |
| 37 | + const sources = this.sources; |
| 38 | + const len = sources.length; |
| 39 | + |
| 40 | + const context = { completed: 0, total: len, values: emptyArray(len), selector: resultSelector }; |
| 41 | + for (let i = 0; i < len; i++) { |
| 42 | + let source = sources[i]; |
| 43 | + if (isPromise(source)) { |
| 44 | + source = new PromiseObservable(<Promise<any>>source); |
| 45 | + } |
| 46 | + (<Observable<any>>source).subscribe(new AllSubscriber(subscriber, i, context)); |
| 47 | + } |
| 48 | + } |
| 49 | +} |
| 50 | + |
| 51 | +class AllSubscriber<T> extends Subscriber<T> { |
| 52 | + private _value: any = null; |
| 53 | + |
| 54 | + constructor(destination: Subscriber<any>, |
| 55 | + private index: number, |
| 56 | + private context: { completed: number, |
| 57 | + total: number, |
| 58 | + values: any[], |
| 59 | + selector: (...values: Array<any>) => any }) { |
| 60 | + super(destination); |
| 61 | + } |
| 62 | + |
| 63 | + _next(value: any): void { |
| 64 | + this._value = value; |
| 65 | + } |
| 66 | + |
| 67 | + _complete(): void { |
| 68 | + const destination = this.destination; |
| 69 | + |
| 70 | + if (this._value == null) { |
| 71 | + destination.complete(); |
| 72 | + } |
| 73 | + |
| 74 | + const context = this.context; |
| 75 | + context.completed++; |
| 76 | + context.values[this.index] = this._value; |
| 77 | + const values = context.values; |
| 78 | + |
| 79 | + if (context.completed !== values.length) { |
| 80 | + return; |
| 81 | + } |
| 82 | + |
| 83 | + if (values.every(hasValue)) { |
| 84 | + let value = context.selector ? context.selector.apply(this, values) : |
| 85 | + values; |
| 86 | + destination.next(value); |
| 87 | + } |
| 88 | + |
| 89 | + destination.complete(); |
| 90 | + } |
| 91 | +} |
| 92 | + |
| 93 | +function hasValue(x: any): boolean { |
| 94 | + return x !== null; |
| 95 | +} |
| 96 | + |
| 97 | +function emptyArray(len: number): any[] { |
| 98 | + let arr = []; |
| 99 | + for (let i = 0; i < len; i++) { |
| 100 | + arr.push(null); |
| 101 | + } |
| 102 | + return arr; |
| 103 | +} |
0 commit comments