I am doing Coroutines-Rx2 interop. According to <https://github.com/Kotlin/kotlinx.coroutines/blob/m...

# coroutines

[aaverin]

I am doing Coroutines-Rx2 interop.
According to https://github.com/Kotlin/kotlinx.coroutines/blob/master/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt

@ExperimentalCoroutinesApi
public fun <T : Any> Deferred<T>.asSingle(context: CoroutineContext): Single<T> = GlobalScope.rxSingle(context) {
    [e-mail]()
}

calling myDeferred.asSingle(coroutineContext) will immediately call await() on the deferred, which effectively starts coroutine execution.
Shouldn’t this only happen when Single will be subscribed on?

[aaverin]

Imagine you have a network call behind coroutine.
I wouldn’t want it to be executed every time your method is called, only when it’s been subscribed on

[bezrukov]

@aaverin CoroutineScope.rxSingle extension creates new single, which does nothing without subscription

[aaverin]

would mySingle.toObservable() subscribe then?

[bezrukov]

nope, only if mySingle.toObservable().subscribe(), mySingle.toObservable() is just a cold observable

[aaverin]

When writing this

println("outside")
            async {
                println("inside")
            }.asSingle(CommonPool)

I get 2 prints
inside and outside

[aaverin]

there should only be outside, until I subscribe

[simon.vergauwen]

That behaviour is encoded here. https://github.com/Kotlin/kotlinx.coroutines/blob/a97fd43b9d79fdef9e1d6a270d6e097089432e26/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt#L32

[simon.vergauwen]

The problem with the code you provided is that async runs eagerly, so the println is printed eagerly. That behavior comes from async and cannot be made lazy by Single.

[simon.vergauwen]

println("outside")
async(start = CoroutineStart.LAZY) {
  println("inside")
}.asSingle(CommonPool)

This should yield your expected result.

[simon.vergauwen]

So only if the Deferred is lazy than the Single will also be lazy. Otherwise Single just `await`s the eagerly running Deferred.

[aaverin]

Wow, this is one nasty pitfall when using both coroutines and rx2 in the same project.
Thanks

[simon.vergauwen]

You're very welcome :) I'd avoid Deferred as return types and use suspend fun instead. Discouples the implementation of the execution strategy.

[aaverin]

What if I depend on the library that returns Deferred?

[aaverin]

I need to somehow convert it to Rx, but if internally async is not Lazy – all execution will be made right away, without subscription

[simon.vergauwen]

Create an intermediate suspend fun in between. You need a layer of in-direction in between.

[simon.vergauwen]

import kotlinx.coroutines.Deferred
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async

object ExternalApi {
  fun helloWorld(): Deferred<Unit> = GlobalScope.async {
    println("Hello World!")
  }
}

suspend fun indirection(): Unit =
  ExternalApi.helloWorld().await()

[simon.vergauwen]

The suspend fun should only be invoked on subscription, this will cause a new deferred to start running.

[simon.vergauwen]

This requires that helloWorld returns a new Deferred every-time, i.e a new network request every-time and not some cached internal values.

[simon.vergauwen]

So still not a fool proof solution tho. Luckily most frameworks like Retrofit now have support for suspend functions instead of Deferred.

[aaverin]

Yes, the full example would be something like

fun convertToProperSingle(): Single<ExternalResponse> {
  return async(start = Coroutine.LAZY) { callExternalApi() }
}.asSingle(CommonPool)

private suspend fun callExternalApi(): ExternalResponse {
  return externalApi.call().await()
}

[simon.vergauwen]

Is there no way to go from suspend fun to Single without Deferred?

[aaverin]

Let me see. I think not

[simon.vergauwen]

We have an API for this in Arrow. Similar to the snippet below

fun <A> asSingle(ctx: CoroutineContext, f: suspend () -> A): Single<A> = Single.create { emitter ->
  f.startCoroutine(Continuation(ctx) { result ->
     result.fold({ a -> emitter.onSuccess(a) }, { e -> emitter.onFailure(e) })
  })
}

[simon.vergauwen]

This is written from my mind so I am not 100% sure about the emitter#xxxx function names.

[simon.vergauwen]

fun convertToProperSingle(): Single<ExternalResponse> =
 asSingle(CommonPool) { externalApi.call().await() }

[aaverin]

public fun <T> Deferred<T>.asSingle

[aaverin]

it’s an extension function on Deferred

[simon.vergauwen]

This is your snippet updated with the conversion function I provided. There should be something similar in kotlinx tho but I’d have to check.

[aaverin]

But yes, maybe will work with your snippet

[aaverin]

thanks

[simon.vergauwen]

Any time 🙂

[Zach Klippenstein (he/him) [MOD]]

CoroutineScope.rxSingle is already lazy, just use it directly instead of going through the asSingle extension.

[Zach Klippenstein (he/him) [MOD]]

fun convertToProperSingle(): SingleExternalResponse {
  return GlobalScope.rxSingle(CommonPool) {
    callExternalApi()
  }
}

private suspend fun callExternalApi(): ExternalResponse {
  return externalApi.call().await()
}