javascript - Cách triển khai observable nóng hết hạn thời gian trong RxJS (hoặc nói chung trong Reactive Extensions)

Tôi muốn triển khai bộ nhớ đệm hết hạn theo thời gian với RxJs. Đây là ví dụ về bộ nhớ đệm "bình thường":

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

Khi subscribe trên cachedData được gọi lần đầu tiên, "nhiệm vụ nặng" sẽ được gọi, và sau 2 giây kết quả được lưu trong cachedData (AsyncSubject). Bất kỳ lần subscribe nào sau đó trên cachedData sẽ trả về ngay lập tức với kết quả đã lưu (do đó là triển khai bộ nhớ đệm).

Điều tôi muốn đạt được là 'làm mới' điều này với khoảng thời gian mà cachedData còn hiệu lực, và khi thời gian đó trôi qua, tôi muốn chạy lại 'công việc nặng' để lấy dữ liệu mới và lưu lại vào cache cho khoảng thời gian mới, v.v...

Hành vi mong muốn:

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

Tôi mới làm quen với Rx(Js) và không thể tìm ra cách triển khai observable nóng này kèm theo thời gian hồi.

Câu trả lời #1

Điều duy nhất bạn thiếu là lập lịch một tác vụ để thay thế cachedData của bạn bằng một AsyncSubject mới sau một khoảng thời gian. Đây là cách thực hiện nó như một phương thức Rx.Observable mới:

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

Cách sử dụng :

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

Câu trả lời #2

Một giải pháp đơn giản là tạo một toán tử có thể nối ống tùy chỉnh để repeatWhen sau khi một khoảng thời gian đã trôi qua. Đây là những gì tôi nghĩ ra:

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

Sau đó tôi sử dụng nó như thế này:

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

Lưu ý quan trọng: Điều này sử dụng publishReplay(1), refCount() vì shareReplay(1) không hủy đăng ký từ observable gốc nên nó sẽ tiếp tục truy cập máy chủ của bạn mãi mãi. Thật không may, điều này dẫn đến việc khi có lỗi, lỗi đó sẽ được phát lại từ publishReplay(1), refCount(). Sẽ có một phiên bản shareReplay 'mới cải tiến' sắp ra mắt. Xem các ghi chú tại đây về một câu hỏi tương tự. Khi phiên bản 'mới' này có sẵn, câu trả lời này nên được cập nhật - nhưng điểm hay của các toán tử tùy chỉnh là bạn có thể sửa chúng ở một nơi duy nhất.