javascript - Kako implementirati hot observable s istekom vremena u RxJS-u (ili općenito u Reactive Extensions)

Volio bih implementirati cache s vremenskim istekom koristeći RxJs. Evo primjera "normalnog" cachea:

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

Kada se subscribe na cachedData pozove prvi put, poziva se „težak posao“, i nakon 2 sekunde rezultat se sprema u cachedData (AsyncSubject). Svaki naredni subscribe na cachedData odmah vraća spremljeni rezultat (tako implementacija predmemorije).

Ono što bih želio postići je "začiniti" ovo s vremenskim periodom tijekom kojeg je cachedData važeći, i kada to vrijeme prođe, želio bih ponovno pokrenuti "heavy duty job" za nove podatke i opet ih spremiti u cache za novi vremenski period, itd...

Željeno ponašanje:

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

Nov sam u Rx(Js) i ne mogu shvatiti kako implementirati ovaj hot observable s razdobljem hlađenja.

Odgovor #1

Sve što vam preostaje je zakazati zadatak za zamjenu vašeg cachedData s novim AsyncSubject nakon određenog vremenskog razdoblja. Evo kako to učiniti kao novu Rx.Observable metodu:

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

Upotreba :

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

Odgovor #2

Jednostavno rješenje bilo bi stvoriti prilagođeni operator koji se može koristiti u lancu za repeatWhen nakon što protekne određeno vrijeme. Evo što sam smislio:

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

Tada ga koristim ovako:

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

Važna napomena: Ovo koristi publishReplay(1), refCount() jer shareReplay(1) ne odjavljuje se s izvornog observabla, pa će stalno slati zahtjeve na vaš server. Nažalost, to ima posljedicu da će se u slučaju pogreške ta pogreška reproducirati iz publishReplay(1), refCount(). Uskoro dolazi 'nova poboljšana' verzija shareReplay. Pogledajte napomene ovdje na sličnom pitanju. Kada ova 'nova' verzija bude dostupna, ovaj odgovor bi trebao biti ažuriran - ali ljepota prilagođenih operatora je u tome što ih možete popraviti na jednom mjestu.