rxjava，如何检查 Single 的结果

问题描述

使用kotlin，有代码

fun fetchRemoteDataApi(): Single<RemoteDataResponse> = networkApi.getData()

    // it is just a retrofit
    @GET(".../api/getData")
    fun getData() : Single<RemoteDataResponse>

fun mergeApiWithDb(): Completable = fetchRemoteDataApi()
            .zipWith(localDao.getAll())
            .flatMapCompletable { (remoteData,localData) ->
                doMerge(remoteData,localData) //<== return a Completable
            }

代码流程：

    val mergeApiDbCall = mergeApiWithDb().onErrorComplete().cache() //<=== would like do some inspection at this level
    PublishSubject.create<Unit>().toFlowable(BackpressureStrategy.LATEST)
                .compose(Transformers.flowableIO())
                .switchMap {
                    //merge DB with api,or local default value first then listen to DB change
                    mergeApiDbCall.andThen(listAllTopics())
                            .concatMapSingle { topics -> remoteTopicUsers.map { topics to it } }
                }
                .flatMapCompletable { (topics,user) ->
                    // do something return Completable
                }
                .subscribe({
                    ...
                },{ throwable ->
                    ...
                })

以及拨打电话时

val mergeApiDbCall = mergeApiWithDb().onErrorComplete().cache()

问题是是否要检查从 Singles<RemoteDataResponse> 返回的 fetchRemoteDataApi()（即使用 Log.i(...) 打印出 RemoteDataResponse 的内容等） ，无论是出现错误还是成功的情况下，怎么办？

/// the functions
fun listAllTopics(): Flowable<List<String>> = localRepoDao.getAllTopics()

// which a DAO:
    @Query("SELECT topic FROM RemoteDataTable WHERE read = 1")
    fun getAllTopics(): Flowable<List<String>>

///
private val remoteTopicUsers: Single<List<User>>
        get() {
            return Single.create {
                networkApi.getTopicUsers(object : ICallback.IGetTopicUsersCallback {
                    override fun onSuccess(result: List<User>) = it.onSuccess(result)
                    override fun onError(errorCode: Int,errorMsg: String?) = it.onError(Exception(errorCode,errorMsg))
                })
            }
        }

解决方法

您无法从 Completable 中提取有关元素的信息。尽管您可以在 doOnComplete() 上使用 Completable，但它不会为您提供有关该元素的任何信息。

如果您在 doOnSuccess() 上调用 Single，您可以检查元素，因此您需要更早地将此调用合并到您的代码中。要检查错误，您可以在 doOnError() 或 Completable 上使用 Single。