RxJava building cache with observables for android

Question

I'm having a hard time understanding on how to build a cache using RxJava. The idea is that I need to either get data from a memory cache or load from my database (dynamoDb). However, this cache is supposed to be shared across fragments and or threads. So I need to return existing observables that are currently running and not finished. This allows threads to catch up and not do unnecessary work. I'm new to RxJava so this is what I have in mind as a sketch (missing some code for brevity):

public class DBCache<K, T> {
private final ConcurrentHashMap<K, Set<T>> resultCache = new ConcurrentHashMap<>;
private final ConcurrentHashMap<K, Observable<Set<T>>> observableCache = new ConcurrentHashMap<>;

private Observable<Set<T>> getFromCache(final DynamoDbCacheKey<K, T> query) {
    return Observable.create(new Observable.OnSubscribe<Set<T>>() {
        @Override
        public void call(Subscriber<? super Set<T>> subscriber) {
        Set<T> results = resultCache.get(query.getKey());
        if (results != null && results.size() > 0) {
            subscriber.onNext(results);
        }
        subscriber.onCompleted();
        }
    });
}

public Observable<Set<T>> get(final QueryCacheKey<K, T> query){
    Observable<Set<T>> cachedObservable = observableCache.get(query.getKey());
    if (cachedObservable != null) {
        return cachedObservable;
    }
    Observable<Set<T>> observable = Observable
        .concat(getFromCache(query), getFromNetwork(query))
        .first()
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .cache();
    observableCache.putIfAbsent(query.getKey(), observable);
return observable;
} 

private Observable<Set<T>> getFromNetwork(final QueryCacheKey<K, T> query) {
    return Observable.create(new Observable.OnSubscribe<Set<T>>() {
    @Override
    public void call(Subscriber<? super Set<T>> subscriber) {
        try {
            Set<T> results = loadFromDb(query); //omitted
            resultCache.putIfAbsent(query.getKey(), results);
            subscriber.onNext(results);
            subscriber.onCompleted();
            observableCache.remove(query.getKey());
        } catch (Exception exception) {
            subscriber.onError(exception);
        }
    }
    });
} 

}

Is there a better way to achieve this through RxJava (Not interested in caching strategies). Any thoughts?

Answer 1

Here is a simple example of doing caching and retrieving a value once:

public class RxCache<K, V> {

    final ConcurrentHashMap<K, AsyncSubject<V>> cache;

    final Func1<K, Observable<V>> valueGenerator;

    public RxCache(Func1<K, Observable<V>> valueGenerator) {
        this.valueGenerator = valueGenerator;
        this.cache = new ConcurrentHashMap<>();
    }

    public Observable<V> get(K key) {
        AsyncSubject<V> o = cache.get(key);
        if (o != null) {
            return o;
        }

        o = AsyncSubject.create();

        AsyncSubject<V> p = cache.putIfAbsent(key, o);
        if (p != null) {
            return p;
        }

        valueGenerator.call(key).subscribe(o);

        return o;
    }

    public void remove(K key) {
        cache.remove(key);
    }
}

If you have multiple values, replace AsyncSubject with ReplaySubject.