It drives me crazy. Rx version works and Coroutine...

# coroutines

[ghedeon]

It drives me crazy. Rx version works and Coroutine version is flaky and failing in 25% of the runs. Do I miss anything about GlobalScope.rxFlowable{}?

GlobalScope_rxFlowable__.kt

[gildor]

Probably coroutine version uses coroutine dispatcher with event loop, so event is not always available, add await for value before check

[gildor]

Bot sure how rxFlowable implemented, but all dispatchers, even runBlocking and Unconfined use event loop now

[gildor]

Probably in this case rxFlowable running on Default dispatcher

[ghedeon]

yes, my guess was about dispatchers and wrapping in runBlocking didn't help

[gildor]

runBlocking also has event loop of events, so it asyncronous even on the same thread

[gildor]

Use some of await functions of TestObserver to await event

[ghedeon]

yes, that works

[ghedeon]

the real case is a bit more complicated, tho. I don't have access to the flowable itself, because it's being converted to LiveData (don't ask why 😒) LiveDataReactiveStreams.fromPublisher(flowable). So, what ends up in the public API is the livedata, backed by this flowable.

[ghedeon]

Thus, somewhere in the test I have liveData.observeForever(observer) and naturally, the observer is flaky.

[ghedeon]

@Test
    fun foo() {
        // given
        val call: () -> Unit = mock()

        val flowable = GlobalScope.rxFlowable {
            call()
            send("Foo")
        }
        val liveData = LiveDataReactiveStreams.fromPublisher(flowable)

        // when
        liveData.observeForever(mock())

        // then
        verify(call).invoke()
    }

[ghedeon]

Any idea how to make this stuff sequential, assuming that flowable is not exposed?

[ghedeon]

I guess, I should inject dispatcher in the rxFlowable then...

[elizarov]

GlobalScope.rxFlowable(Dispatchers.Unconfined)

[ghedeon]

Hm.. it becomes sequential with rxFlowable(Dispatchers.Unconfined)

[elizarov]

That is the equivalent of Rx version 😉

[ghedeon]

simple smile

[elizarov]

Rx is unconfined by default, Coroutines work in background by default. But you can change defaults in both libs.

[ghedeon]

Thank you! What about GlobalScope.rxFlowable{}.subscribeOn(scheduler)? Shouldn't provided rx scheduler somehow replace the coroutine dispatcher? Doesn't seem like it's happening now.

[elizarov]

subscribeOn only works for sources that do not have their own scheduler.