Publish with multiple consumers doesn't work as expected

[sskrla]

This may be a misunderstanding on my part, but coming from using Reactor (and in fact, porting code written in Reactor) I would expect the following flow to pass:

    @Test
    fun `publish multiple consumers`() = runBlocking {
        arrayOf(1, 2, 3, 4, 5)
                .asFlow()
                .publish {
                    shared ->
                        merge(shared.filter { it % 2 == 1}, shared.filter { it % 2 == 0})
                }
                .assertResult(1, 2, 3, 4, 5)
    }

But in reality, the flow completes yielding no results. Using flatMapConcat will yield results from the first sequence, and not the second.

[akarnokd]

What do you mean by yielding no results? Does the assertion fail?

[sskrla]

Updated the code with a test I've added to the project. It fails with:

expected:<[1, 2, 3, 4, 5]> but was:<[]>
Expected :[1, 2, 3, 4, 5]
Actual   :[]

[akarnokd]

For some reason, merge doesn't start collecting the inner subject and the main source just runs to completion without observers. I have no idea why that is.

I know they had quite some trouble with their shareIn operator too. Kotlin/kotlinx.coroutines#2047

[akarnokd]

This is a complicated issue and I'm not sure how to solve it without asking the programmer for more parameters.

The workaround I suggest is in the form of an overload to publish where you can tell it to wait for these 2 inner collectors.

Closing via adec011