javascript - Kuinka toteuttaa ajan umpeutumisen hot-observable RxJS:ssä (tai yleisesti Reactive Extensionsissa)

Haluaisin toteuttaa Aikakatkaisu-välimuistin RxJs:llä. Tässä on esimerkki "tavallisesta" välimuistista:

//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"*/ });
});

Kun subscribe kutsutaan cachedData -kohteella ensimmäisen kerran, kutsutaan "heavy duty job", ja 2 sekunnin kuluttua tulos tallennetaan cachedData -kohteeseen (AsyncSubject). Kaikki myöhemmät subscribe-kutsut cachedData-kohteeseen palauttavat heti tallennetun tuloksen (näin ollen välimuistin toteutus).

Haluaisin saavuttaa sen, että lisään vähän maustetta siihen ajanjaksoon, jonka cachedData on voimassa, ja kun tuo aika kuluu, haluaisin suorittaa uudelleen "raskaan tehtävän" uuden datan varten ja välimuistittaa tämän uudelleen uudelle ajanjaksolle jne...

Toivottu käyttäytyminen:

//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

Olen uusi Rx(Js):ssä enkä pysty selvittämään, kuinka toteuttaa tämä kuuma havaittava kohteen jäähdytyksellä.

Vastaus #1

Ainoa, mikä puuttuu, on ajoittaa tehtävä, joka korvaa sinun cachedData-tietosi uudella AsyncSubject-kohteella tietyn ajanjakson jälkeen. Tässä on, kuinka tehdä se uutena Rx.Observable-metodina:

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

Käyttö :

// 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();

Vastaus #2

Yksinkertainen ratkaisu olisi luoda mukautettu putkikelpoinen operaattori repeatWhen, kun tietty aika on kulunut. Tässä on mitä keksin:

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

Sitten käytän sitä näin:

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

Tärkeä huomautus: Tämä käyttää publishReplay(1), refCount() koska shareReplay(1) ei lopeta tilausta alkuperäisestä observablesta, joten se jatkaa palvelimesi kuormittamista loputtomasti. Valitettavasti tämä aiheuttaa sen, että virheen sattuessa kyseinen virhe toistetaan publishReplay(1), refCount()-kohdasta. Pian tulee saataville 'uusi parannettu' shareReplay. Katso lisätiedot täältä samankaltaisesta kysymyksestä. Kun tämä 'uusi' versio on saatavilla, tähän vastaukseen tulisi tehdä päivitys - mutta mukavaa mukautetuissa operaattoreissa on se, että voit korjata ne yhdessä paikassa.