Caching network calls using RxJava for some duration

Question

I am making a network using Retorfit + RxJava2 and I want to cache the response for 30 seconds. Any calls made after 30 seconds interval should get the latest results from server. I tried doing this using Replay operator but it still makes a network call every time I call subscribe. I am not an expert in RxJava so maybe my understanding of using Replay for caching like that is wrong.

public Observable<Name> getName() {
        return retrofitBuilder.getName()
                .subscribeOn(Schedulers.io())
                .replay(30, TimeUnit.SECONDS,Schedulers.io())
                .autoConnect();
    }

and I am calling the above code like this:

 service.getName()
        .subscribe(new Consumer<Name>()
            {
                @Override
                public void accept(Name name) throws Exception
                {
                    Log.d("getName", "Name: " + name.toString());
                }
            }
            , new Consumer<Throwable>()
            {
                @Override
                public void accept(Throwable throwable) throws Exception
                {
                    Log.d("getName", throwable.getMessage());
                }
            });

UPDATE: My apology if I didn't explain my question clearly. What I want is caching on a particular request instead of caching it on HttpClient level which applies the caching strategy to all the request being made through it. In the end I would like to define different caching expiration for different request when needed. Not all my request needs caching for small duration. I was wondering if I could do just that.

Appreciate your help in this.

Contradance

each call to service.getName() will return a new Observable. That replay operator, and others like it, are meant to allow multicasting of various sorts on a single Observable. Most likely, you'd want to implement caching behind the service.getName() method, such that it will offer caching functionality regardless of how many separate Observables you create that run it. This might mean you implement caching at the Http client layer, with "Cache-Control", or some other method. – Denitadenitrate

One solution would be to use a different okhttp client/interceptor for that call only. – Stearin

Answer 1

The are 2 problem with your approach:

1. as @drhr mentioned, you are creating a new Observable each time you call service.getName() you're creating a new instance of Observable, you should keep the same replayed instance and give to the caller outside the same instance each time it calls service.getName().
2. even if you will return the same instance, replay with 30 seconds, will replay the sequence emitted by the source Observable over the last 30 sec, meaning after cache expiration time, you will get nothing as your request happened more than 30 sec ago. it doesn't mean that the Observable will restart automatically after this period.

In order to cache for specific period, you basically need to invalidate the cached response after cache period, and perform new request after this period, that's mean you should control your subscribe, and do it there.
You can achieve it with something like that:

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

with defer you can return the right Observable according to cache expiration status, so every subscribe happened within the cache expiration will get cached Observable (using cache()) - meaning request will be performed only once. after cache expiration, additional subscribe will trigger new request and will set a new timer to reset the cache expiration.

Uncloak

This solution seems pretty awesome. Let me try that and get back to you. Thanks for responding. – Contradance

I tried your solution but it is not working. I debugged it and it seems it is taking the cached observable but still making the network request. Please help. ` public class UserNameService { CachedRequest<String> obs; public Observable<String> getUserName() { if (obs == null) { Observable<String> observable = RetrofitBuilder.getInstance().groupList(); obs = new CachedRequest<>(observable, 1, TimeUnit.MINUTES); } return obs.getCachedObservable() .subscribeOn(Schedulers.io()); } } – Contradance

how do you use it exactly? the creation of the CachedObservable is not guarded against multithreaded, but i doubt if that's the problem in your case – Uncloak

I am using exactly the same way I have shown in above comment. I am getting an observable from the network request, passing it to CacheRequest constructor the first time when I create an instance and call getCachedObservable. Then next time I use this cachedRequest member variable instance(NOT create a new one) and calling getCachedObservable. The caller of this method is just getting the observable and subscribing to it to display the message on View. Can you share how you suggest to use it? – Contradance

Answer 2

Try to look at okhttp interceptors.

Add CacheInterceptor:

public class CacheInterceptor implements Interceptor {
    @Override
    public Response intercept(Chain chain) throws IOException {
        Response response = chain.proceed(chain.request());

        CacheControl cacheControl = new CacheControl.Builder()
                .maxAge(30, TimeUnit.SECONDS)
                .build();

        return response.newBuilder()
                .removeHeader("Pragma")
                .removeHeader("Cache-Control")
                .header("Cache-Control", cacheControl.toString())
                .build();
    }
}

And add it and cache to your OkHttp Client like this:

File httpCacheDirectory = new File(context.getCacheDir(), "http-cache");
int cacheSize = 10 * 1024 * 1024; // 10 MiB
Cache cache = new Cache(httpCacheDirectory, cacheSize);

OkHttpClient httpClient = new OkHttpClient.Builder()
                               .addNetworkInterceptor(new CacheInterceptor())
                               .cache(cache)
                               .build();

Roundy

It seems this will cache all the request made by this builder for 30 seconds. Also, if I need different expiration for different request, I have to create multiple OkHttpClient in this case. It will be nice if I can just cache a particular request with cache expiration. – Contradance

@vab It seems like you haven't written about this new conditions – Roundy

I have made some amendments in my question as you suggested. Thanks for pointing out. – Contradance