Loading data from Database + Network (Room + Retrofit + RxJava2)

Question

I have a sample API request which returns a list of user's watchlist. I want to achieve the following flow when the user loads the watchlist screen:

1. Load the data from DB cache immediately.(cacheWatchList)

2. Initiate the RetroFit network call in the background.

   i. onSuccess return apiWatchList
   ii. onError return cacheWatchList

3. Diff cacheWatchList vs apiWatchList

   i. Same -> all is well since data is already displayed to the user do nothing.

   ii. Differs -> Save apiWatchList to a local store and send the apiWatchList to the downstream.

What I have done so far?

Watchlist.kt

data class Watchlist(
  val items: List<Repository> = emptyList()
)

LocalStore.kt (Android room)

  fun saveUserWatchlist(repositories: List<Repository>): Completable {    
    return Completable.fromCallable {      
      watchlistDao.saveAllUserWatchlist(*repositories.toTypedArray())
    }
  }

RemoteStore.kt (Retrofit api call)

  fun getWatchlist(userId: UUID): Single<Watchlist?> {
    return api.getWatchlist(userId)
  }

DataManager.kt

  fun getWatchlist(userId: UUID): Flowable<List<Repository>?> {
    val localSource: Single<List<Repository>?> =
      localStore.getUserWatchlist()
        .subscribeOn(scheduler.computation)

    val remoteSource: Single<List<Repository>> = remoteStore.getWatchlist(userId)
      .map(Watchlist::items)
      .doOnSuccess { items: List<Repository> ->
        localStore.saveUserWatchlist(items)
          .subscribeOn(scheduler.io)
          .subscribe()
      }
      .onErrorResumeNext { throwable ->
        if (throwable is IOException) {
          return@onErrorResumeNext localStore.getUserWatchlist()
        }
        return@onErrorResumeNext Single.error(throwable)
      }
      .subscribeOn(scheduler.io)

    return Single.concat(localSource, remoteSource)
  }

The problem with the above flow is, it calls onNext twice for each stream source to the downstream(presenter) even though both the data are same.

I can do the data diff logic in the presenter and update accordingly but I want the DataManager class to handle the logic for me(CleanArchitecture, SOC).

My Questions?

1. What's the best possible way to implement the above logic?

2. Am I leaking the inner subscriptions in DataManager (see: doOnSuccess code) ?. I'm disposing of the outer subscription when the presenter is destroyed.

Answer 1

fun getWatchlist(userId: UUID): Observable<List<Repository>?> {
val remoteSource: Single<List<Repository>> = 
remoteStore.getWatchlist(userId)
        .map(Watchlist::items)
        .subscribeOn(scheduler.io)

return localStore.getUserWatchlist()
        .flatMapObservable { listFromLocal: List<Repository> ->
            remoteSource
                    .observeOn(scheduler.computation)
                    .toObservable()
                    .filter { apiWatchList: List<Repository> ->
                        apiWatchList != listFromLocal
                    }
                    .flatMapSingle { apiWatchList ->
                        localSource.saveUserWatchlist(apiWatchList)
                                .andThen(Single.just(apiWatchList))
                    }
                    .startWith(listFromLocal)
        }
}

Explanation step by step:

1. Load data from localStore
2. Use flatMapObservable to subscribe to remoteSource each time the localStore emits data.
3. As there are more than one emission from inner observable(initial data from local and new data in case of updated data from the remoteSource) transform Single to Observable.
4. Compare data from remoteSource with data from the localStore and proceed data only in case if newData != localData.
5. For each emission after the filter initiate the localSource to save data and on a completion of this operation proceed saved data as Single.
6. As requested, at the beginning of remote request data from localStore should be proceeded and it is simply done be adding startWith at the end of the operators chain.