RxJava kotlin flatmap 链网络协程返回结果

Question

首先，我以前从未使用过 RxJava。对于所提供的所有用例和示例，我必须实现一个使用 AndroidRx 或 RxJava 的 SDK。

我有以下方法从后端返回数据

override suspend fun getSomeData(token: String): Outcome<String, DataError> {
        val getSomeDataMutation = GetSomeDataMutation(token)

        try {
            val response = apolloClient.mutate(getSomeDataMutation).await()

            return if (response.data?.someData != null) {
                Outcome.success(response.data!!.someData)
            } else {
                Outcome.error(DataError.AN_ERROR(R.string.error))
            }
        } catch(...)

现在我正在尝试实现如下所示的方法，以使用 RxJava 与 SDK 交互，这将在 flatmap 中调用上述方法（基于示例）。 这甚至可能吗？ 我已经尝试了几个小时，但没有任何成功。如果我修改上述方法以返回 SingleSource，它会起作用，但我不能这样做。

TheSDK.requestToken()
            .rx()
            .flatMap { token ->
                network.getSomeData(token)  //this part is causing the problem
            }
            .flatMap { networkResult->
                return@flatMap TheSDK.session(SomeResponse.with(apiResult)).rx()
            }
            .subscribe { sessResult, throwable ->
                if (sessResult!= null) {
                    if (sessResult.isSuccessful) {
                        // success
                    } else {
                        // something wrong
                    }
                } else {
                    //an error occurred during the stream
                }
            }

Answer 1

我认为你使用了错误的方法。

当您尝试链接 2 个反应性事物时，您应该使用 .flatMap

 fun myFun(): Single<String> {
    val intSingle: Single<Int> = Single.just(1)
    val stringSingle: Single<String> = Single.just("foo")

    return intSingle
        .flatMap { stringSingle }
}

另一方面，当您尝试将反应性事物与非反应性事物链接时，您应该使用.map

 fun myFun(): Single<String> {
    val intSingle: Single<Int> = Single.just(1)
    val string: String = "foo"

    return intSingle
        .map { string }
}

查看您的代码，我相信解决方案是：

TheSDK.requestToken()
        .rx()
        .map{ token ->
            network.getSomeData(token)  // returns something non-reactive
        }
        .flatMap { networkResult->
            TheSDK.session(SomeResponse.with(apiResult))
                .rx()
        }
        .subscribe(
            { result ->
                // do something onSuccess
            },
            { error ->
                // do something onError
            }
        )

如您所见，我还使用了一个订阅者，您可以在其中传递 2 个 lambda，一个 onSuccess 和 onError