Pause/resume semantics on an Observable.

pausable.js

    /**
     * Pauses the underlying observable sequence based upon the observable sequence which yields true/false.
     * @example
     * var pauser = new Rx.Subject();
     * var source = Rx.Observable.interval(100).pausable(pauser);
     * @param {Observable} pauser The observable sequence used to pause the underlying sequence.
     * @returns {Observable} The observable sequence which is paused based upon the pauser.
     */
    observableProto.pausable = function (pauser) {
        var self = this;
        return observableCreate(function (observer) {
            var conn = self.publish(),
                subscription = conn.subscribe(observer),
                connection = disposableEmpty;

            var pausable = pauser.distinctUntilChanged().subscribe(function (b) {
                if (b) {
                    connection = conn.connect();
                } else {
                    connection.dispose();
                    connection = disposableEmpty;
                }
            });
        });
    };

usage.js

var pauser = new Rx.Subject();

var source = Rx.Observable.interval(1000).pausable(pauser);

var subscription = source.subscribe(
    function (x) {
        console.log('onNext: ' + x);
    },
    function (err) {
        console.log('onError: ' + err);
    },
    function () {
        console.log('onCompleted');
    }
);

var isRunning = true;

setInterval(function () {
   pauser.onNext(isRunning);
   isRunning = !isRunning;
}, 5000);

// => next: 0
// => next: 1
// => next: 2
// => next: 3
// => next: 0
// => next: 1
// => next: 2
// => next: 3
// => next: 0
// => next: 1
// => next: 2
// => next: 3

So basically this is great for a lossy implementation. Something along the lines of sample. But what if we didn't want to resubscribe? What if we wanted the underlying observable/producer to actually pause?