Как реализовать время, которое можно наблюдать в RxJS (или вообще в реактивных расширениях)

Question

Я хотел бы реализовать кеш времени истечения срока действия с 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 называется в первый раз, «сверхмощный работа» называется, и через 2 секунды Результат сохраняется в 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) и не могу понять, как реализовать эту горячий наблюдаемый с кулдауном.

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