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);
    });
};

Here is a working typescript version, in case anyone needs it:

import { Observable } from 'rxjs/Observable';
import { Scheduler as IScheduler } from 'rxjs/Scheduler';
import { AsyncSubject, Scheduler } from 'rxjs/Rx';

declare module 'rxjs/Observable' {
    interface Observable<T> {
        cacheWithExpiration: (expirationMs: number, scheduler?: IScheduler) => Observable<T>;
    }
}

Observable.prototype.cacheWithExpiration = function (expirationMs: number, scheduler?: IScheduler) {
    var cachedData: AsyncSubject<any>,
        source = this;

    scheduler = scheduler || Scheduler.async;

    return Observable.create(observer => {
        if (!cachedData) {
            cachedData = new AsyncSubject();
            source.subscribe(cachedData);
            cachedData.subscribe(data => {
                scheduler.schedule(() => {
                    cachedData = null;
                }, expirationMs);
            });
        }

        return cachedData.subscribe(observer);
    });
}

and in case anyone needs it compile correctly in v6, then try this one:

import { asyncScheduler, AsyncSubject, Observable, SchedulerLike } from 'rxjs';

export const fromCacheWithExperation = <T>(expirationMs: number, scheduler: SchedulerLike = asyncScheduler) => (
    source: Observable<T>,
) => {
    let cached: AsyncSubject<T> | null;
    return new Observable<T>((observer) => {
        if (!cached) {
            cached = new AsyncSubject();
            source.subscribe(cached);
            cached.subscribe(() => {
                scheduler.schedule(() => {
                    cached = null;
                }, expirationMs);
            });
        }
        return cached.subscribe(observer);
    });
};