Sha256: a877ff0de4f2da46cf352be0698efe03be29b7593a83d0019920e48f9a93fd62
Contents?: true
Size: 1.93 KB
Versions: 14
Compression:
Stored size: 1.93 KB
Contents
/**
* Merges an observable sequence of observable sequences into an observable sequence, limiting the number of concurrent subscriptions to inner sequences.
* Or merges two observable sequences into a single observable sequence.
*
* @example
* 1 - merged = sources.merge(1);
* 2 - merged = source.merge(otherSource);
* @param {Mixed} [maxConcurrentOrOther] Maximum number of inner observable sequences being subscribed to concurrently or the second observable sequence.
* @returns {Observable} The observable sequence that merges the elements of the inner sequences.
*/
observableProto.merge = function (maxConcurrentOrOther) {
if (typeof maxConcurrentOrOther !== 'number') { return observableMerge(this, maxConcurrentOrOther); }
var sources = this;
return new AnonymousObservable(function (o) {
var activeCount = 0, group = new CompositeDisposable(), isStopped = false, q = [];
function subscribe(xs) {
var subscription = new SingleAssignmentDisposable();
group.add(subscription);
// Check for promises support
isPromise(xs) && (xs = observableFromPromise(xs));
subscription.setDisposable(xs.subscribe(function (x) { o.onNext(x); }, function (e) { o.onError(e); }, function () {
group.remove(subscription);
if (q.length > 0) {
subscribe(q.shift());
} else {
activeCount--;
isStopped && activeCount === 0 && o.onCompleted();
}
}));
}
group.add(sources.subscribe(function (innerSource) {
if (activeCount < maxConcurrentOrOther) {
activeCount++;
subscribe(innerSource);
} else {
q.push(innerSource);
}
}, function (e) { o.onError(e); }, function () {
isStopped = true;
activeCount === 0 && o.onCompleted();
}));
return group;
}, sources);
};
Version data entries
14 entries across 7 versions & 1 rubygems