javascript - Hvordan implementerer man en tidsudløbs hot observable i RxJS (eller generelt i Reactive Extensions)

Jeg vil gerne implementere Time Expiry-cache med RxJs. Her er et eksempel på en "normal" cache:

//let this represents "heavy duty job"
var data = Rx.Observable.return(Math.random() * 1000).delay(2000);

//and we want to cache result
var cachedData = new Rx.AsyncSubject();
data.subscribe(cachedData);

cachedData.subscribe(function(data){
    //after 2 seconds, result is here and data is cached
    //next subscribe returns immediately data
    cachedData.subscribe(function(data2){ /*this is "instant"*/ });
});

Når subscribe på cachedData kaldes første gang, bliver "heavy duty job" kaldt, og efter 2 sekunder gemmes resultatet i cachedData (AsyncSubject). Enhver anden efterfølgende subscribe på cachedData returnerer straks med det gemte resultat (altså cache-implementering).

Hvad jeg gerne vil opnå, er at "krydder" dette med den tidsperiode, hvor cachedData er gyldig, og når den tid er gået, vil jeg gerne køre "tungt job" igen for nye data og cache dette igen for en ny tidsperiode osv...

Ønsket adfærd:

//pseudo code
cachedData.youShouldExpireInXSeconds(10);


//let's assume that all code is sequential from here

//this is 1.st run
cachedData.subscribe(function (data) {
    //this first subscription actually runs "heavy duty job", and
    //after 2 seconds first result data is here
});

//this is 2.nd run, just after 1.st run finished
cachedData.subscribe(function (data) {
    //this result is cached
});

//15 seconds later
// cacheData should expired
cachedData.subscribe(function (data) {
    //i'm expecting same behaviour as it was 1.st run:
    // - this runs new "heavy duty job"
    // - and after 2 seconds we got new data result
});


//....
//etc

Jeg er ny til Rx(Js) og kan ikke finde ud af, hvordan jeg implementerer dette hot observable med cooldown.

Svar #1

Det eneste, du mangler, er at planlægge en opgave for at erstatte din cachedData med en ny AsyncSubject efter en tidsperiode. Her er, hvordan du gør det som en ny Rx.Observable-metode:

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

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

    return Rx.Observable.create(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 = undefined;
                });
            });
        }

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

Brug:

// a *cold* observable the issues a slow query each time it is subscribed
var data = Rx.Observable.return(42).delay(5000);

// the cached query
var cachedData = data.cacheWithExpiration(15000);

// first observer must wait
cachedData.subscribe();

// wait 3 seconds

// second observer gets result instantly
cachedData.subscribe();

// wait 15 seconds

// observer must wait again
cachedData.subscribe();

Svar #2

En enkel løsning ville være at oprette en brugerdefineret pipebar operator til repeatWhen, at en varighed er passeret. Her er, hvad jeg kom op med:

 export const refreshAfter = (duration: number) => (source: Observable<any>) =>
                                  source.pipe(
                                     repeatWhen(obs => obs.pipe(delay(duration))),
                                     publishReplay(1), 
                                     refCount());

Så bruger jeg det sådan her:

const serverTime$ = this.environmentClient.getServer().pipe(map(s => s.localTime))
const cachedServerTime$ = serverTime.pipe(refreshAfter(5000)); // 5s cache

Vigtig bemærkning: Dette bruger publishReplay(1), refCount() fordi shareReplay(1) ikke afmelder sig fra kilde-observable, så det vil fortsætte med at ramme din server for evigt. Desværre har dette den konsekvens, at ved fejl vil den fejl blive gengivet fra publishReplay(1), refCount(). Der kommer snart en 'ny forbedret' shareReplay. Se noter her om et lignende spørgsmål. Når denne 'nye' version er tilgængelig, bør dette svar opdateres - men det smarte ved brugerdefinerede operatører er, at du kan rette dem ét sted.