Convert RXJava Single to a coroutine's Deferred?

user_1060472 0
Viewed 8416

I have a Single from RxJava and want to continue working with a Deferred from Kotlin Coroutines. How to accomplish that?

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

I would be interested in some library (if there is any?) as well as in doing this on my own... So far I did this hand-made implementation on my own:

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
}
rx-java kotlin-coroutines kotlin coroutine
3 Answers

There is this library that integrates RxJava with Coroutines: https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2

There's no function in that library to directly convert a single to a Deferred though. The reason for this is probably that an RxJava Single is not bound to a coroutine scope. If you want to convert it to a Deferred you would therefore need to provide it a CoroutineScope.

You could probably implement it like this:

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

The Single.await function (used in the async-block) is from the kotlinx-coroutines-rx2 library.

You can call the function like this:

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

Just turn your Single into a suspend function:

suspend fun waitForRxJavaResult(): String? = suspendCoroutine { cont ->
        try {
            val result = resultSingle.blockingGet()
            cont.resume(result)
        } catch (e: Exception){
            cont.resume(null)
        }
    }
user_74517790 
implementation 'io.reactivex.rxjava2:rxandroid:+'
implementation 'io.reactivex.rxjava2:rxjava:+'
implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-rx2:+'

And in code add:

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

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

Usage:

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



 fun getItemMaybe(): Maybe<Int> {
       
 }
user_24258510
Related