kotlinx-coroutines-rx3: Allow specifing CoroutineStart for rxSingle, rxObservable, ...

[Gabweb]

Description

Coroutines (e.g. launch) offer a parameter to specify the CoroutineStart. This is a request to add the same start parameter to the rx migration libraries (kotlinx-coroutines-rx2 and kotlinx-coroutines-rx3).

Use case

Consider the following two code examples - one in pure RxJava and one while migrating from RxJava to Coroutines.

RxJava3

println("Start")
Single.just("Rx").subscribe { value -> println(value)  }
println("End")

// Prints Start, Rx, End 

RxJava3 + Coroutines

println("Start")
rxSingle { "Coroutine" }.subscribe { value -> println(value)  }
println("End")

// Prints Start, End, Coroutine 

While the code looks nearly identical, they behave differently (which itself is fine). However, the current API of e.g. rxSingle does not allow specifying the CoroutineStart parameter. Therefore, it's impossible to achieve the same start / scheduling behavior and makes migration difficult.

Side-note: Using Dispatcher.Confined resolves the timing difference. However, it will also overwrite the Dispatcher and might change the Thread, which is not desired.

The Shape of the API

fun <T : Any> rxSingle(
     context: CoroutineContext = EmptyCoroutineContext,
     start: CoroutineStart = CoroutineStart.DEFAULT,
     block: suspend CoroutineScope.() -> T
): Single<T>

Usage example:

println("Start")
rxSingle(start = CoroutineStart.UNDISPATCHED) { "Coroutine" }.subscribe { value -> println(value)  }
println("End")

// Prints Start, Coroutine, End 

(This should also apply to rxObservable, ...)

Prior Art

Under-the-hood, rxSingle und others are currently hard-coded to CoroutineStart.Default. To enable this feature, the parameter just needs to be exposed publicly.

[dkhalanskyjb]

I don't understand a reason to replace Single.just(value) with rxSingle { value }. Why start a coroutine at all if the value is immediately available? And if the value is not available, then the proposed solution won't help anyway, as the thread will switch to Dispatchers.Default as soon as there is even a single suspension in the code block, so it's still impossible to achieve the same scheduling behavior.

Could you describe your migration process in more detail? Maybe with some specific code snippets: what's the original snippet, what should it look like after the migration, and what intermediate steps would the new start parameter allow that are impossible today?