How to pass Observable emissions to MutableSharedFlow?

user_4252667 0
Viewed 235

well, I have an Observable, I’ve used asFlow() to convert it but doesn’t emit. I’m trying to migrate from Rx and Channels to Flow, so I have this function

override fun processIntents(intents: Observable<Intent>) {
     intents.asFlow().shareTo(intentsFlow).launchIn(this)
}

shareTo() is an extension function which does onEach { receiver.emit(it) }, processIntents exists in a base ViewModel, and intentsFlow is a MutableSharedFlow.

fun <T> Flow<T>.shareTo(receiver: MutableSharedFlow<T>): Flow<T> {
    return onEach { receiver.emit(it) }
}

I want to pass emissions coming from the intents Observable to intentsFlow, but it doesn’t work at all and the unit test keeps failing.

@Test(timeout = 4000)
fun `WHEN processIntent() with Rx subject or Observable emissions THEN intentsFlow should receive them`() {
    return runBlocking {

        val actual = mutableListOf<TestNumbersIntent>()
        val intentSubject = PublishSubject.create<TestNumbersIntent>()


        val viewModel = FlowViewModel<TestNumbersIntent, TestNumbersViewState>(
            dispatcher = Dispatchers.Unconfined,
            initialViewState = TestNumbersViewState()
        )

        viewModel.processIntents(intentSubject)


        intentSubject.onNext(OneIntent)
        intentSubject.onNext(TwoIntent)
        intentSubject.onNext(ThreeIntent)

        viewModel.intentsFlow.take(3).toList(actual)

        assertEquals(3, actual.size)
        assertEquals(OneIntent, actual[0])
        assertEquals(TwoIntent, actual[1])
        assertEquals(ThreeIntent, actual[2])
    }
}

test timed out after 4000 milliseconds org.junit.runners.model.TestTimedOutException: test timed out after 4000 milliseconds

rx-java kotlin-flow kotlin-coroutines kotlin
1 Answers

This works

val ps = PublishSubject.create<Int>()
val mf = MutableSharedFlow<Int>()
val pf = ps.asFlow()
    .onEach {
        mf.emit(it)
    }
launch {
    pf.take(3).collect()
}
launch {
    mf.take(3).collect {
        println("$it") // Prints 1 2 3
    }
}
launch {
    yield() // Without this we suspend indefinitely
    ps.onNext(1)
    ps.onNext(2)
    ps.onNext(3)
}

We need the take(3)s to make sure our program terminates, because MutableSharedFlow and PublishSubject -> Flow collect indefinitely.

We need the yield because we're working with a single thread and we need to give the other coroutines an opportunity to start working.

Take 2

This is much better. Doesn't use take, and cleans up after itself.

After emitting the last item, calling onComplete on the PublishSubject terminates MutableSharedFlow collection. This is a convenience, so that when this code runs it terminates completely. It is not a requirement. You can arrange your Job termination however you like.

Your code never terminating is not related to the emissions never being collected by the MutableSharedFlow. These are separate concerns. The first is due to the fact that neither a flow created from a PublishSubject, nor a MutableSharedFlow, terminates on its own. The PublishSubject flow will terminate when onComplete is called. The MutableSharedFlow will terminate when the coroutine (specifically, its Job) collecting it terminates.

The Flow constructed by PublishSubject.asFlow() drops any emissions if, at the time of the emission, collection of the Flow hasn't suspended, waiting for emissions. This introduces a race condition between being ready to collect and code that calls PublishSubject.onNext().

This, I believe, is the reason why flow collection isn't picking up the onNext emissions in your code.

It's why a yield is required right after we launch the coroutine that collects from psf.

val ps = PublishSubject.create<Int>()
val msf = MutableSharedFlow<Int>()
val psf = ps.asFlow()
    .onEach {
        msf.emit(it)
    }
val j1 = launch {
    psf.collect()
}

yield() // Use this to allow psf.collect to catch up

val j2 = launch {
    msf.collect {
        println("$it") // Prints 1 2 3 4
    }
}

launch {
    ps.onNext(1)
    ps.onNext(2)
    ps.onNext(3)
    ps.onNext(4)
    ps.onComplete()
}
j1.invokeOnCompletion { j2.cancel() }
j2.join()
user_3697220
Related