javascript - Cómo implementar un observable caliente con expiración de tiempo en RxJS (o en general en Extensiones Reactivas)

Me gustaría implementar una caché con tiempo de expiración usando RxJs. Aquí hay un ejemplo de una caché "normal":

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

Cuando se llama a subscribe en cachedData por primera vez, se llama a "heavy duty job", y después de 2 segundos el resultado se guarda en cachedData (AsyncSubject). Cualquier otra suscripción posterior a cachedData devuelve inmediatamente el resultado guardado (por lo tanto, implementación de caché).

Lo que me gustaría lograr es "darle sabor" a esto con un período de tiempo durante el cual cachedData es válido, y cuando ese tiempo pase, me gustaría volver a ejecutar el "trabajo pesado" para obtener nuevos datos y almacenarlos en caché nuevamente por un nuevo período de tiempo, etc...

Comportamiento deseado:

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

Soy nuevo en Rx(Js) y no puedo averiguar cómo implementar este observable caliente con tiempo de espera.

Respuesta #1

Lo único que te falta es programar una tarea para reemplazar tu cachedData con un nuevo AsyncSubject después de un período de tiempo. Aquí se explica cómo hacerlo como un nuevo método de 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);
    });
};

Uso :

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

Respuesta #2

Una solución simple sería crear un operador encadenable personalizado para repeatWhen que pase una duración. Esto es lo que se me ocurrió:

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

Entonces lo estoy usando así:

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

Nota importante: Esto utiliza publishReplay(1), refCount() porque shareReplay(1) no se desuscribe del observable de origen, por lo que continuará haciendo peticiones a tu servidor para siempre. Desafortunadamente, esto tiene la consecuencia de que, en caso de error, ese error será reproducido desde el publishReplay(1), refCount(). Pronto habrá un shareReplay 'nuevo y mejorado'. Consulta las notas aquí sobre una pregunta similar. Una vez que esta versión 'nueva' esté disponible, esta respuesta debería actualizarse, pero la belleza de los operadores personalizados es que puedes corregirlos en un solo lugar.