return sourceObservable .replay(1, 30, TimeUnit.SECONDS) .autoConnect() .switchIfEmpty(sourceObservable);

Solicitud inicial: va a la fuente Segunda solicitud dentro de los 30 segundos posteriores a la emisión de la fuente: entregada desde la memoria caché Tercera solicitud fuera de la ventana de caducidad de la memoria caché: nada. Me suscribo y no obtengo datos, pero no cambia a la fuente ascendente Observable.

El problema aquí es que la reproducción simplemente repite la misma secuencia emitida por sourceObservable en los últimos 30 segundos, pero cuando te suscribes después de 30 segundos, la secuencia no tiene eventos, ni siquiera onCompleted() , por lo que no puedes switchIfEmpty() , no funcionará ya que depende de 'onCompleted()' y sin ninguna emisión, para saber que está 'vacío'.

En general, el uso de la reproducción no es suficiente en el escenario de caché, ya que lo que necesita es una forma de volver a suscribirse en caso de que caduque el caché y, además, hacerlo por demanda, es decir, solo cuando algún cliente se suscribe a él. (puede hacer un caché que se actualice cada 30 segundos, pero supongo que ese no es el comportamiento deseado)

Entonces, como sugirió @Yurly Kulikov, debe mantener un estado y controlar la operación de suscripción para mantener el estado. Pero creo que hay un flujo importante en la solución, ya que en realidad no es exactamente seguro para subprocesos, lo que significa que si 2 se suscribe uno tras otro, digamos A y B, mientras que A ejecuta la solicitud y espera para guardar el nuevo como resultado en el caché, B también puede suscribirse, y se ejecutará otra solicitud como valor almacenado en caché que A aún no ha establecido (aún no finalizó la primera solicitud de red).

Sugiero usar un enfoque similar con una implementación diferente, que sugerí aquí :

 public class CachedRequest<T> { private final AtomicBoolean expired = new AtomicBoolean(true); private final Observable<T> source; private final long cacheExpirationInterval; private final TimeUnit cacheExpirationUnit; private Observable<T> current; public CachedRequest(Observable<T> o, long cacheExpirationInterval, TimeUnit cacheExpirationUnit) { source = o; current = o; this.cacheExpirationInterval = cacheExpirationInterval; this.cacheExpirationUnit = cacheExpirationUnit; } private Observable<T> getCachedObservable() { return Observable.defer(() -> { if (expired.compareAndSet(true, false)) { current = source.cache(); Observable.timer(cacheExpirationInterval, cacheExpirationUnit) .subscribe(aLong -> expired.set(true)); } return current; }); } }

con diferir puede devolver el Observable correcto de acuerdo con el estado de caducidad de la memoria caché, por lo que cada suscripción que se produjo dentro de la caducidad de la memoria caché se almacenará en Observable en caché (usando cache()), lo que significa que la solicitud se realizará solo una vez. después de la caducidad de la memoria caché, la suscripción adicional activará una nueva solicitud y establecerá un nuevo temporizador para restablecer la caducidad de la memoria caché.

Tendrá que mantener un estado compartido entre muchas personas que llaman. Es por eso que no puede crear el Observable cada vez que se llama a getContentObservable().

Una forma de hacerlo es crear un Observable afuera, mantener el estado interno en el Observable (por ejemplo, usando un búfer), pero implementar un comportamiento con estado a menudo es más fácil sin Observables.

Aquí hay un ejemplo con estado compartido en un campo:

 private Optional<String> cached = Optional.empty(); private Observable<String> getContentObservable() { //use defer to delay cache evaluation to the point when someone subscribes return Observable.defer( () -> cached.isPresent() ? Observable.just(cached) : fetchAndCache() ) //use the same scheduler for every cached field access .subscribeOn(scheduler); } private Observable<String> fetchAndCache() { Observable<String> cachedSource = getSourceObservable() //I assume you only need one, make sure it is 1 .take(1) .cache(); cachedSource .observeOn(scheduler) //side-effect stores the state .doOnNext(str -> cached = Optional.of(str)) .flatMap(str -> Observable.timer(30, TimeUnit.SECONDS, scheduler)) //another side-effect clears the cache .subscribe(l -> cached = Optional.empty()); return cachedSource; }

Entonces resulta que puede usar el recurso compartido de reproducción de Jake Wharton para almacenar en caché el último valor incluso después de desecharlo. https://github.com/JakeWharton/RxReplayingShare