有没有更好的方法在android中用rxjava实现单源真值

Question

在我的应用程序中，我有使用Room的数据库和使用retrofit的网络服务。我有一个要求，如果本地数据库中没有数据，我需要查询网络并显示一个进度条。如果网络返回空数据，那么我需要显示一个空视图。其中一个问题是，我需要忽略来自房间的空数据，只考虑来自服务器的空数据，这样当用户没有任何数据时，他只是看到一个加载的视图，而在服务器返回空数据后，他将看到空视图。

我是用发布主题来实现的。Lce(loading content error)是围绕数据的包装对象。

val recentPublish = PublishSubject.create<Lce<List<RecentMessage>>>()

fun loadRecentMessages() {
        loadMessageFromDB()
        loadRecentMessageFromServer()
}
private fun loadMessageFromDB() {
    disposable = recentMessageDao.getRecentMessages() // this is a flowable
        .subscribeOn(Schedulers.io())
        .subscribe({
            Timber.d("recent message from db size ${it.size}")
            handleMessageFromDB(it)
        }, {
            it.printStackTrace()
            Timber.e("error on flowable from db!")
        })
}
protected fun handleMessageFromDB(messages: List<RecentMessage>) {
    // only publish if the data is not empty
    if (messages.isNotEmpty()) {
        recentPublish.onNext(Lce.Content(messages))
    }
}

private fun loadRecentMessageFromServer() {
    recentPublish.onNext(Lce.Loading())
    networkService.getLatestMessage() // this is a single
        .subscribe({
            val parsedMessages =
                DtoConverter.convertRecentPrivateMessageResponse(it, user.id!!)
            handleMessageFromServer(parsedMessages)
        }, {
            it.printStackTrace()
            recentPublish.onNext(Lce.Error(it))
            Timber.w("failed to load recent message for private chat from server")
        })
}

private fun handleMessageFromServer(recentMessages: List<RecentMessage>) {
    Timber.i("recent messages from server ${recentMessages.size}")
    if (recentMessages.isEmpty()) {
        recentPublish.onNext(Lce.Content(arrayListOf()))
    } else {
        recentMessageDao.saveAll(recentMessages)
    }
}

在上面的代码中，我只传递来自服务器的空数据，而忽略了来自房间的空数据。这个解决方案是可行的，但我想知道是否有更好的功能方法来解决这个问题。我是Rxjava的初学者，希望得到任何帮助。谢谢你的帮助。

Answer 1

经过一番研究和@EpicPandaForce的评论，我想到了这个方法。我学到了不少东西，它只是点击了我，关于如何正确使用rxjava。这是我的方法，任何评论都将感激不尽。

    fun getMessages(): Observable<Lce<List<RecentMessage>>> {
        return Observable.mergeDelayError(getMessagesFromDB(), getMessagesFromNetwork()) // even if network fails, we still want to observe the DB
    }

    private fun getMessagesFromDB(): Observable<Lce.Content<List<RecentMessage>>> {
        return recentMessageDao.getRecentMessages()
            .filter {
                it.isNotEmpty() // only forward the data from db if it's not empty
            }.map {
                Lce.Content(it)
            }
    }

    private fun getMessagesFromNetwork(): Observable<Lce<List<RecentMessage>>> {
        // first show a loading , then request for data
        return Observable.concat(Observable.just(Lce.Loading()), profileService.getLatestMessage()
                .flatMap {
                    processServerResponse(it) // store the data to db
                }.onErrorReturn {
                    Lce.Error(it)
                }.filter {
                    (it as Lce.Content).packet.isEmpty() // only forward data if it's empty
                })
    }


    private fun processServerResponse(response: RecentMessageResponse): Observable<Lce<List<RecentMessage>>> {
        return Observable.create {
            val parsedMessages =
                DtoConverter.convertRecentPrivateMessageResponse(response, user.id!!)
            handleMessageFromServer(parsedMessages)
            it.onComplete() // we use single source of truth so don't return anyting
        }
    }