Skip to content

Instantly share code, notes, and snippets.

@jhusain
Created July 10, 2014 01:21
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jhusain/a70692de3366e111f6b4 to your computer and use it in GitHub Desktop.
Save jhusain/a70692de3366e111f6b4 to your computer and use it in GitHub Desktop.
/*
* Performs a exclusive map waiting for the first to finish before subscribing to another observable.
* Observables that come in between subscriptions will be dropped on the floor.
* @param {Function} selector Selector to invoke for every item in the current subscription.
* @param {Any} [thisArg] An optional context to invoke with the selector parameter.
* @returns {Observable} An exclusive observable with only the results that happen when subscribed.
*/
observableProto.exclusiveMap = function (selector, thisArg) {
var sources = this;
return new AnonymousObservable(function (observer) {
var index = 0,
hasCurrent = false,
isStopped = true,
m = new SingleAssignmentDisposable(),
g = new CompositeDisposable();
g.add(m);
m.setDisposable(sources.subscribe(
function (innerSource) {
if (!hasCurrent) {
hasCurrent = true;
innerSubscription = new SingleAssignmentDisposable();
g.add(innerSubscription);
isPromise(innerSource) && (innerSource = observableFromPromise(innerSource));
innerSubscription.setDisposable(innerSource.subscribe(
function (x) {
var result;
try {
result = selector.call(thisArg, x, index++, innerSource);
} catch (e) {
observer.onError(e);
return;
}
observer.onNext(result);
}
},
observer.onError.bind(observer),
function () {
g.remove(innerSubscription);
hasCurrent = false;
if (isStopped && g.length === 1) {
observer.onCompleted();
}
}))
}
},
observer.onError.bind(observer),
function () {
isStopped = true;
if (g.length === 1 && !hasCurrent) {
observer.onCompleted();
}
}));
return g;
});
};
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment