我想用RxJs实现Time Expiry缓存.以下是“普通”缓存的示例:
//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"*/ });
});
当第一次调用对cachedData的订阅时,将调用“重载作业”,并在2秒后将结果保存在cachedData(AsyncSubject)中. 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)的新手,无法弄清楚如何用冷却来实现这个热观察.
解决方法:
您缺少的是在一段时间后安排任务用新的AsyncSubject替换您的cachedData.以下是如何将其作为新的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();