Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
/*
* Performs a exclusive waiting for the first to finish before subscribing to another observable.
* Observables that come in between subscriptions will be dropped on the floor.
* @returns {Observable} A exclusive observable with only the results that happen when subscribed.
*/
observableProto.exclusive = function () {
var sources = this;
return new AnonymousObservable(function (observer) {
var hasCurrent = false,
isStopped = false,
m = new SingleAssignmentDisposable(),
g = new CompositeDisposable();
g.add(m);
m.setDisposable(sources.subscribe(
function (innerSource) {
if (!hasCurrent) {
hasCurrent = true;
isPromise(innerSource) && (innerSource = observableFromPromise(innerSource));
var innerSubscription = new SingleAssignmentDisposable();
g.add(innerSubscription);
innerSubscription.setDisposable(innerSource.subscribe(
observer.onNext.bind(observer),
function(e) {
hasCurrent = false;
observer.onError(e);
},
function () {
g.remove(innerSubscription);
hasCurrent = false;
if (isStopped && g.length === 1) {
observer.onCompleted();
}
));
}
},
observer.onError.bind(observer),
function () {
hasCurrent = false;
isStopped = true;
if (g.length === 1) {
observer.onCompleted();
}
}));
return g;
});
};
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.