javascript - Bagaimana untuk melaksanakan hot observable yang mempunyai tamat masa dalam RxJS (atau secara amnya dalam Reactive Extensions)

Saya ingin melaksanakan cache Tamat Masa dengan RxJs. Berikut adalah contoh cache "biasa":

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

Apabila subscribe pada cachedData dipanggil untuk pertama kali, "tugas berat" dipanggil, dan selepas 2 saat keputusan disimpan dalam cachedData (AsyncSubject). Mana-mana subscribe seterusnya pada cachedData akan kembali serta-merta dengan keputusan yang disimpan (dengan itu pelaksanaan cache).

Apa yang saya ingin capai adalah untuk 'menambah perisa' dengan tempoh masa dalam cachedData adalah sah, dan apabila tempoh itu berlalu, saya ingin menjalankan semula 'tugas berat' untuk data baru dan menyimpannya dalam cache sekali lagi untuk tempoh masa baru, dan sebagainya...

Tingkah laku yang diingini:

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

Saya baru dengan Rx(Js) dan tidak dapat mencari cara untuk melaksanakan observable panas ini dengan tempoh sejuk.

Jawapan #1

Yang anda tinggal hanyalah menjadualkan tugasan untuk menggantikan cachedData anda dengan AsyncSubject baru selepas satu tempoh masa. Berikut ialah cara untuk melakukannya sebagai kaedah Rx.Observable baru:

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

Penggunaan :

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

Jawapan #2

Satu penyelesaian yang mudah ialah dengan mencipta operator boleh-paip tersuai untuk repeatWhen setelah suatu tempoh masa berlalu. Berikut adalah apa yang saya hasilkan:

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

Kemudian saya menggunakannya seperti ini:

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

Nota penting: Ini menggunakan publishReplay(1), refCount() kerana shareReplay(1) tidak berhenti melanggan dari observable sumber jadi ia akan terus menghubungi pelayan anda untuk selamanya. Malangnya, ini mempunyai akibat bahawa apabila berlaku ralat, ralat itu akan dimainkan semula dari publishReplay(1), refCount(). Akan datang terdapat versi shareReplay 'baru dan diperbaiki'. Lihat nota di sini mengenai soalan yang serupa. Setelah versi 'baru' ini tersedia, jawapan ini harus dikemas kini - tetapi keindahan operator tersuai ialah anda boleh membaikinya di satu tempat sahaja.