/* * 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, selectorFunc = bindCallback(selector, thisArg, 3); 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 = selectorFunc(x, index++, innerSource); } catch (e) { observer.onError(e); return; } observer.onNext(result); }, function (e) { observer.onError(e); }, function () { g.remove(innerSubscription); hasCurrent = false; if (isStopped && g.length === 1) { observer.onCompleted(); } })); } }, function (e) { observer.onError(e); }, function () { isStopped = true; if (g.length === 1 && !hasCurrent) { observer.onCompleted(); } })); return g; }, this); };