جافاسكريبت - كيفية تنفيذ كائن قابل للمراقبة ذو انتهاء وقت في RxJS (أو بشكل عام في Reactive Extensions)

أود تنفيذ ذاكرة التخزين المؤقت بانتهاء المدة مع RxJs. إليك مثال على ذاكرة التخزين المؤقت "العادية":

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

عندما يتم استدعاء subscribe على cachedData لأول مرة، يتم استدعاء "المهمة الثقيلة"، وبعد ثانيتين يتم حفظ النتيجة في cachedData (AsyncSubject). أي subscribe آخر لاحق على cachedData يُرجع على الفور مع النتيجة المحفوظة (وبالتالي تنفيذ التخزين المؤقت).

ما أود تحقيقه هو إضافة بعض "البهارات" لهذا بحيث تكون فترة الوقت ضمن cachedData صالحة، وعندما تنقضي تلك الفترة، أود إعادة تشغيل "الوظيفة الشاقة" للحصول على بيانات جديدة وتخزينها مؤقتًا مرة أخرى لفترة جديدة، وهكذا...

السلوك المرغوب:

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

أنا جديد على Rx(Js) ولا أستطيع معرفة كيفية تنفيذ هذا الـ observable الساخن مع فترة التهدئة.

الإجابة رقم 1

كل ما تنقصه هو جدولة مهمة لاستبدال cachedData الخاص بك بـ AsyncSubject جديد بعد فترة زمنية. إليك كيفية القيام بذلك كطريقة جديدة لـ 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);
    });
};

الاستخدام :

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

الإجابة رقم 2

حل بسيط سيكون إنشاء عامل قابل للأنابيب مخصص لـ repeatWhen بعد مرور مدة معينة. إليك ما توصلت إليه:

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

ثم أستخدمه هكذا:

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

ملاحظة مهمة: هذا يستخدم publishReplay(1)، refCount() لأن shareReplay(1) لا يقوم بإلغاء الاشتراك من المراقب الأصلي لذا سيستمر في الوصول إلى الخادم الخاص بك إلى الأبد. ولسوء الحظ، فإن لهذا العيب أنه عند حدوث خطأ، سيتم إعادة تشغيل هذا الخطأ من publishReplay(1) وrefCount(). هناك نسخة 'جديدة ومحسنة' من shareReplay ستأتي قريبًا. انظر الملاحظات هنا حول سؤال مشابه. بمجرد توفر هذه النسخة 'الجديدة' يجب تحديث هذا الجواب - لكن جمال المشغلات المخصصة هو أنه يمكنك إصلاحها في مكان واحد.