This repository has been archived by the owner on Apr 20, 2018. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 2.1k
/
combinelatest.js
94 lines (81 loc) · 3.38 KB
/
combinelatest.js
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
function falseFactory() { return false; }
function argumentsToArray() {
var len = arguments.length, args = new Array(len);
for(var i = 0; i < len; i++) { args[i] = arguments[i]; }
return args;
}
var CombineLatestObservable = (function(__super__) {
inherits(CombineLatestObservable, __super__);
function CombineLatestObservable(params, cb) {
this._params = params;
this._cb = cb;
__super__.call(this);
}
CombineLatestObservable.prototype.subscribeCore = function(observer) {
var len = this._params.length,
subscriptions = new Array(len);
var state = {
hasValue: arrayInitialize(len, falseFactory),
hasValueAll: false,
isDone: arrayInitialize(len, falseFactory),
values: new Array(len)
};
for (var i = 0; i < len; i++) {
var source = this._params[i], sad = new SingleAssignmentDisposable();
subscriptions[i] = sad;
isPromise(source) && (source = observableFromPromise(source));
sad.setDisposable(source.subscribe(new CombineLatestObserver(observer, i, this._cb, state)));
}
return new NAryDisposable(subscriptions);
};
return CombineLatestObservable;
}(ObservableBase));
var CombineLatestObserver = (function (__super__) {
inherits(CombineLatestObserver, __super__);
function CombineLatestObserver(o, i, cb, state) {
this._o = o;
this._i = i;
this._cb = cb;
this._state = state;
__super__.call(this);
}
function notTheSame(i) {
return function (x, j) {
return j !== i;
};
}
CombineLatestObserver.prototype.next = function (x) {
this._state.values[this._i] = x;
this._state.hasValue[this._i] = true;
if (this._state.hasValueAll || (this._state.hasValueAll = this._state.hasValue.every(identity))) {
var res = tryCatch(this._cb).apply(null, this._state.values);
if (res === errorObj) { return this._o.onError(res.e); }
this._o.onNext(res);
} else if (this._state.isDone.filter(notTheSame(this._i)).every(identity)) {
this._o.onCompleted();
}
};
CombineLatestObserver.prototype.error = function (e) {
this._o.onError(e);
};
CombineLatestObserver.prototype.completed = function () {
this._state.isDone[this._i] = true;
this._state.isDone.every(identity) && this._o.onCompleted();
};
return CombineLatestObserver;
}(AbstractObserver));
/**
* Merges the specified observable sequences into one observable sequence by using the selector function whenever any of the observable sequences or Promises produces an element.
*
* @example
* 1 - obs = Rx.Observable.combineLatest(obs1, obs2, obs3, function (o1, o2, o3) { return o1 + o2 + o3; });
* 2 - obs = Rx.Observable.combineLatest([obs1, obs2, obs3], function (o1, o2, o3) { return o1 + o2 + o3; });
* @returns {Observable} An observable sequence containing the result of combining elements of the sources using the specified result selector function.
*/
var combineLatest = Observable.combineLatest = function () {
var len = arguments.length, args = new Array(len);
for(var i = 0; i < len; i++) { args[i] = arguments[i]; }
var resultSelector = isFunction(args[len - 1]) ? args.pop() : argumentsToArray;
Array.isArray(args[0]) && (args = args[0]);
return new CombineLatestObservable(args, resultSelector);
};