我想用 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"*/ });
});
When subscribe
on cachedData
第一次调用,调用“重载作业”,2秒后结果保存在cachedData
(AsyncSubject
)。任何其他后续subscribe
on 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) 的新手,无法弄清楚如何通过冷却来实现这个热可观察值。
您所缺少的只是安排一个任务来取代您的cachedData
与一个新的AsyncSubject
一段时间后。以下是如何作为新人进行操作Rx.Observable
method:
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);
});
};
Usage:
// 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();
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)