Issue with getting cached then fresh data

[jamieadkins95]

We are using the recipe from the wiki to retrieve cached then fresh data:

store.get(barCode)
                .concatWith(store.fetch(barCode))
                .distinct()

Under normal conditions this works fine. When fresh data request times out, we get any cached data and then onError is called later when the request times out.

However I'm running into an issue when the device is in airplane mode. In this case the fetch call fails immediately with an UnknownHostException, onError is then called and the whole observable chain collapses. Since the failure is immediate, the observable chain collapses before the cached data can be returned. Instead of getting cached data and an error message, we get just an error message.

How should I handle this case? Should I be just checking for a network connection before making the request, or is there an approach I can take with Store to address this?

For reference here is how we are currently using Store:

fun <T> getData(barCode: BarCode, source: Single<ResponseBody>, type: Type = genericType<Class<T>>(), expirationTime: Long = 5): Flowable<T> {
        var store = storeMap[barCode] as? Store<T, BarCode>
        if (store == null) {
            val storeBuilder = StoreBuilder.parsedWithKey<BarCode, BufferedSource, T>()
                    .fetcher { source.map(ResponseBody::source) }
                    .memoryPolicy(MemoryPolicy
                            .builder()
                            .setExpireAfterWrite(expirationTime)
                            .setExpireAfterTimeUnit(TimeUnit.SECONDS)
                            .build())
                    .parser(GsonParserFactory.createSourceParser<T>(ApiManager.getGson(), type))
            try {
                storeBuilder.persister(FileSystemRecordClearingPersister.create(
                        FileSystemFactory.create(CoreApplication.instance.cacheDir),
                        BarCodePathResolver(),
                        expirationTime,
                        TimeUnit.SECONDS))
            } catch (e: Exception) {
                Timber.e(e, "Failed to get file store")
            }

            store = storeBuilder.open()
            storeMap.put(barCode, store)
        }

        return store.get(barCode)
                .concatWith(store.fetch(barCode))
                .distinct()
    }

[jamieadkins95]

I can work around this particular issue by checking the network connection before calling fetch. e.g.

val activeNetwork = (CoreApplication.instance.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager).activeNetworkInfo
val isConnected = activeNetwork != null && activeNetwork.isAvailable

val cached = store.get(barCode)

return if (isConnected) {
    cached.concatWith(store.fetch(barCode)).distinct()
} else {
    cached.toFlowable()
}

However the point of my question still stands, how can I ensure that cached data is returned when the fetch observable chain collapses?

[ferrytan]

i have the same scenario. my workaround is to use onErrorResumeNext on fetch call and hand it back to get() on error.

store.fetch(barCode).onErrorResumeNext(throwable -> store.get(barcode)

edit: it's actually already featured on one of the issue here.
good to know i use the same method

[digitalbuddha]

Sorry I'm a bit confused. My understanding is that get would be called first returning your cached data and only after that emission completes your fetch would be called. Could you provide a test case showing what your problem is? I don't see how network would be called before cache in the above example. Maybe a regression happened.

[jamieadkins95]

Actually yeah you are right, it shouldn't happen. I didn't realise concat() waited for the first observable to complete. That makes this behaviour very strange.

I'll put together a small test project to try and reproduce it.

[jamieadkins95]

@digitalbuddha

I have a demo project for you to reproduce the bug. See here.

Strangely, I found that I got the expected behaviour (cached data and then onError) if I added a doOnSuccess call after the Store.get()

Doesn't work:

return store.get(barCode)
    .concatWith(store.fetch(barCode))
    .distinct()

Works:

return store.get(barCode)
    .doOnSuccess { Timber.d("onSuccess") }
    .concatWith(store.fetch(barCode))
    .distinct()

Maybe there is some rx behaviour here that I don't understand.

Update:
That doOnSuccess workaround only sometimes works.

[jamieadkins95]

@digitalbuddha Hey, have you had a chance to take a look at this?

[jamieadkins95]

The issue is definitely caused by store.get() completing before the cached data has been completely propagated to the Observer. If you add a 100ms delay before you call store.fetch(), this issue does not occur:

return store.get(barCode)
    .concatWith(
        Completable.timer(100, TimeUnit.MILLISECONDS)
            .andThen(store.fetch(barCode))
    )
    .distinct()

[brianPlummer]

@jamieadkins95 hey there! i was able to replicate your issue and found that if you set the .networkBeforeStale() on the store builder it will function as you want. The internal logic deals with a single and either emits a value or errors. Will consider alternate approaches. Here is the line of code if you are interested: https://bit.ly/2KP3DFc Also, my apologies on getting back to you. Please let us know if this works. Thank you!

[jamieadkins95]

Thanks for taking a look @brianPlummer!

We have usecases where we would like the cached data and also the error so that we can show some data and then an error message to the user. Adding .networkBeforeStale() hides that error.

Our latest implementation has moved away from using onError and now propagates the error through onNext

Something vaguely like this:

val local = store.get(barCode)
    .map { Result.Success(it) }
    .onErrorReturn {  Result.Failure(it) }

val remote = store.fetch(barCode)
    .map { Result.Success(it) }
    .onErrorReturn {  Result.Failure(it) }

return local
    .concatWith(remote)
    .distinct()

This avoids the chain terminating if there is an error fetching fresh data