JavaScript concatLatest operator for RxJS

rx-concatLatest.js

Rx.Observable.prototype.concatLatest = function () {
    /// <summary>
    /// Concatenates an observable sequence of observable sequences, skipping sequences that arrive while the current sequence is being observed.
    /// If N new observables arrive while the current observable is being observed, the first N-1 new observables will be thrown
    /// away and only the Nth will be observed.
    /// </summary>
    /// <returns type="Rx.Observable"></returns>
    var source = this;

    return Rx.Observable.create(function (observer) {
        var latest,
            isStopped,
            isBusy,
            outerSubscription = new Rx.SingleAssignmentDisposable(),
            innerSubscription = new Rx.SerialDisposable(),
            subscriptions = new Rx.CompositeDisposable(outerSubscription, innerSubscription),
            onError = observer.onError.bind(observer),
            onNext = observer.onNext.bind(observer),
            innerOnComplete = function () {
                var inner = latest;
                if (inner) {
                    latest = undefined;
                    innerSubscription.setDisposable(inner.subscribe(onNext, onError, innerOnComplete));
                }
                else {
                    isBusy = false;
                    if (isStopped) {
                        observer.onCompleted();
                    }
                }
            };

        outerSubscription.setDisposable(source.subscribe(function (newInner) {
            if (isBusy) {
                latest = newInner;
            }
            else {
                isBusy = true;
                innerSubscription.setDisposable(newInner.subscribe(onNext, onError, innerOnComplete));
            }
        }, onError, function () {
            isStopped = true;
            if (!isBusy) {
                observer.onCompleted();
            }
        }));

        return subscriptions;
    });
};