RxJs extension implementation for cache results with time expiration

Rx.Observable.cacheWithExpiration-demo.js

var slowJob = Rx.Observable.defer(function () {
    return Rx.Observable.return(Math.random() * 1000).delay(2000);
});

var cached = slowJob.cacheWithExpiration(5000);

var last = Date.now();
function repeat() {
    last = Date.now();
    cached.subscribe(function (data) {
        console.log("number:", data, 'took:', Date.now() - last, '[ms]');
        setTimeout(repeat, 1000);
    });
}
setTimeout(repeat, 1000);

Rx.Observable.cacheWithExpiration.js

Rx.Observable.prototype.cacheWithExpiration = function (expirationMs, scheduler) {
    var source = this;
    var cachedData = null;

    // Use timeout scheduler if scheduler not supplied
    scheduler = scheduler || Rx.Scheduler.timeout;

    return Rx.Observable.createWithDisposable(function (observer) {

        if (!cachedData) {
            // The data is not cached.
            // create a subject to hold the result
            cachedData = new Rx.AsyncSubject();

            // subscribe to the query
            source.subscribe(cachedData);

            // when the query completes, start a timer which will expire the cache
            cachedData.subscribe(function () {
                scheduler.scheduleWithRelative(expirationMs, function () {
                    // clear the cache
                    cachedData = null;
                });
            });
        }

        // subscribe the observer to the cached data
        return cachedData.subscribe(observer);
    });
};