将 RXJava Single 转换为协程的 Deferred？

Question

我有一个来自 RxJava 的 Single 并且想继续使用一个来自 Kotlin Coroutines 的 Deferred。如何实现？

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 的函数。原因可能是 RxJava Single 没有绑定到协程范围。如果要将其转换为 Deferred，则需要为其提供 CoroutineScope.

你可以这样实现它：

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

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

您可以这样调用函数：

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

Answer 2

把你的 Single 变成一个挂起函数：

suspend fun waitForRxJavaResult(): String?{
    return suspendCoroutine { cont ->
        try {
            val result = resultSingle.blockingGet()
            cont.resume(result)
        }catch (e: Exception){
            cont.resume(null)
        }
    }
}

Answer 3

implementation 'io.reactivex.rxjava2:rxandroid:+'
implementation 'io.reactivex.rxjava2:rxjava:+'
implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-rx2:+'

并在代码中添加：

fun <T> Maybe<T>.toDeferred(coroutineScope: CoroutineScope) = coroutineScope.async { awaitSingleOrNull() }

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

用法：

lifecycleScope.launch {
            val result = getItemMaybe().toDeferred(this).await()
            ...
        }



 fun getItemMaybe(): Maybe<Int> {
       
 }

将 RXJava Single 转换为协程的 Deferred？

Convert RXJava Single to a coroutine's Deferred?

kotlin

rx-java

coroutine

kotlin-coroutines