如果没有连接，如何从本地获取我的应用程序，如果有，如何从远程获取？

Question

我的MainRepository 是从 API 获取数据并插入数据库，然后显示在 UI 上的。

override fun fetchAll() {
    Observable.fromCallable { local.fetchPosts() }
        .doOnNext {
            remote.fetchPosts().concatMap { posts ->
                local.insert(*posts.toTypedArray())
                Observable.just(posts)
            }
        }
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { outcome.success(it) },
            { error: Throwable -> outcome.failed(error) }
        ).addTo(compositeDisposable)
}

outcome 变量是PublishObject 类型的Response，即Loading、Success 或Failure。

override val outcome = PublishSubject.create<Response<List<Post>>>()

[...]

sealed class Response<T> {
    companion object {
        fun <T> loading(loading: Boolean): Response<T> = Progress(loading)
        fun <T> success(data: T): Response<T> = Success(data)
        fun <T> failure(e: Throwable): Response<T> = Failure(e)
    }

    data class Progress<T>(var loading: Boolean) : Response<T>()
    data class Success<T>(var data: T) : Response<T>()
    data class Failure<T>(var e: Throwable) : Response<T>()
}

它执行一个方法local.fetchPosts()，这是一个负责访问DAO函数的函数。

fun fetchPosts() = database.postDao().fetchAll()

[...]

@Query("SELECT * FROM posts ORDER BY createdAt DESC")
fun fetchAll(): List<Post>

addTo 是Disposable 的扩展：

fun Disposable.addTo(compositeDisposable: CompositeDisposable) {
    compositeDisposable.add(this)
}

我尝试在Observable.fromCallable 之后使用concatMap，但它会直接显示来自API 的数据，而doOnNext 会从数据库中显示，但它不会更新列表，删除已删除的内容远程服务器。

Answer 1

首先您必须了解doOnNext 是Side Effect Operator，以及它的家族，副作用运算符仅预测执行次要操作（例如记录）的排放，它们不会影响流无论如何。

所以，你正在做可观察的

remote.fetchPosts().concatMap { posts ->
            local.insert(*posts.toTypedArray())
            Observable.just(posts)
        }

永远不会开始工作，因为它没有被订阅。 现在您必须根据您想要的行为做出决定，我假设您的用例是：

• 尝试从 API 获取数据。

• 缓存成功，不查询DAO

• 显示它

然后这样的事情会起作用：

remote.fetchPosts()
    // cache the data from remote.
    .doOnNext(posts -> local.insert(*posts.toTypedArray()))
    // if an error happens, use the posts in the DAO.
    .onErrorResumeNext { Observable.fromCallable { local.fetchPosts() } }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    [...]