将 RXJava Single 转换为协程的 Deferred？

Question

我有一个来自 RxJava 的 Single，并希望继续使用来自 Kotlin 的延迟协程。如何做到这一点？

fun convert(data: rx.Single<String>): kotlinx.coroutines.Deferred<String> = ...

我会对一些库（如果有的话？）以及自己做这件事感兴趣......到目前为止，我自己做了这个手工实现：

private fun waitForRxJavaResult(resultSingle: Single<String>): String? {
    var resultReceived = false
    var result: String? = null

    resultSingle.subscribe({
        result = it
        resultReceived = true
    }, {
        resultReceived = true
        if (!(it is NoSuchElementException))
            it.printStackTrace()
    })
    while (!resultReceived)
        Thread.sleep(20)

    return result
}

Answer 1

有一个将 RxJava 与协程集成的库：https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2

该库中没有直接将单个转换为Deferred尽管的函数。其原因可能是 RxJavaSingle未绑定到协程作用域。如果您想将其转换为 a，Deferred则需要为其提供 a CoroutineScope。

你可能可以这样实现它：

fun <T> Single<T>.toDeferred(scope: CoroutineScope) = scope.async { await() }

该Single.await函数（在async-block 中使用）来自 kotlinx-coroutines-rx2 库。

您可以像这样调用该函数：

coroutineScope {
    val mySingle = getSingle()
    val deferred = mySingle.toDeferred(this)
}