javascript - Cum să implementezi un observable fierbinte cu expirarea timpului în RxJS (sau în general în Reactive Extensions)

Aș dori să implementez un cache cu expirare de timp folosind RxJs. Iată un exemplu de cache „normal”:

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

Când subscribe pe cachedData este apelat pentru prima dată, se apelează „heavy duty job” și după 2 secunde rezultatul este salvat în cachedData (AsyncSubject). Orice alt subscribe ulterior pe cachedData returnează imediat cu rezultatul salvat (astfel implementându-se cache-ul).

Ceea ce aș dori să realizez este să „condimentez” acest lucru cu o perioadă de timp în care cachedData este valabil, iar când acea perioadă trece, aș dori să rulez din nou „sarcina grea” pentru date noi și să le memorez din nou pentru o perioadă de timp nouă, etc...

Comportament dorit:

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

Sunt nou în Rx(Js) și nu reușesc să înțeleg cum să implementez acest observable cald cu răcire.

Răspuns #1

Tot ce îți lipsește este să programezi o sarcină pentru a înlocui cachedData cu un nou AsyncSubject după o perioadă de timp. Iată cum să faci acest lucru ca o nouă metodă Rx.Observable:

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

Utilizare :

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

Răspunsul #2

O soluție simplă ar fi să creezi un operator personalizat care poate fi conectat la un pipe pentru a repeatWhen a trecut o anumită durată. Iată ce am creat:

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

Atunci îl folosesc astfel:

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

Notă importantă: Acest lucru folosește publishReplay(1), refCount() deoarece shareReplay(1) nu se dezabonează de la observable-ul sursă, așa că va continua să atingă serverul tău pentru totdeauna. Din păcate, acest lucru are consecința că, în cazul unei erori, acea eroare va fi re-redată de publishReplay(1), refCount(). Urmează să apară un shareReplay „nou și îmbunătățit”. Vezi note aici despre o întrebare similară. Odată ce această versiune „nouă” va fi disponibilă, acest răspuns ar trebui actualizat - dar frumusețea operatorilor personalizați este că poți să îi repari într-un singur loc.