SharedFlow.emit() doesn't wait for a subscriber to complete collecting

[j2esu]

SharedFlow doc says:

For example, the following class encapsulates an event bus that distributes events to all subscribers in a rendezvous manner, suspending until all subscribers process each event

So, with the following code

fun main() = runBlocking {
    val numbers = MutableSharedFlow<Int>()
    GlobalScope.launch {
        delay(1000)
        repeat(3) {
            println("emit $it")
            numbers.emit(it)
        }
    }
    GlobalScope.launch {
        numbers.collect {
            delay(1000)
            println("$it collected")
        }
    }.join()
}

we could expect the following output:

emit 0
0 collected
emit 1
1 collected
emit 2
2 collected

But the actual output is:

emit 0
emit 1
0 collected
emit 2
1 collected
2 collected

Seems like the flow has an extra 1 size buffer and doesn't suspend on first emit() call. Is it a bug, or I'm missing something?

[elizarov]

Thanks. There's already PR with a fix for the next release: #2437

[serandel]

Sorry for commenting in a closed thread, @elizarov , but I see the PR is just a tweak to the docs. Is there a way to make the SharedFlow suspend till all the subscribers have completed collecting, as we (mis)understood before? For me it would be a very good and easy way of syncronizing the emission of values through several flows.

[j2esu]

Hi, @serandel
Hope that @elizarov has more elegant solution, but for now I have some thoughts about that. We need some sort of "sync" event in our shared flow. It can be null, specific value or type. Then, we can modify an example above:

fun main() = runBlocking {
    val numbers = MutableSharedFlow<Int?>() // UPD: allow null as sync event
    GlobalScope.launch {
        delay(1000)
        repeat(3) {
            println("emit $it")
            numbers.emit(it)
            numbers.emit(null) // UPD: suspend until subscribers receive sync event
            // here we have a guarantee that business event collected, cause sync event received
        }
    }
    GlobalScope.launch {
        numbers.filterNotNull() // UPD: just skip sync event by filtering
            .collect {
                delay(1000)
                println("$it collected")
            }
    }.join()
}

This code will print desired sequential emit-collect output. Of course, we need to use unbuffered shared flow to get suspend behavior.

[serandel]

Thanks, @j2esu!

I really like your solution for SharedFlow but unfortunately it messes up with StateFlow. :/

Right now I'm playing with a decorator like this, but I'm not 100% sure I'm race conditions-free.

class RendezvousSharedFlow<T>(private val flow: MutableSharedFlow<T>) {
    private val rendezvous = Channel<Unit>()

    suspend fun emit(value: T) {
        flow.emit(value)

        repeat(flow.subscriptionCount.value) {
            rendezvous.receive()
        }
    }

    suspend fun collect(collector: suspend (T) -> Unit) {
        flow.collect {
                collector(it)
                rendezvous.send(Unit)
        }
    }
}

[serandel]

I also have to think about managing flows with replay...

[j2esu]

@serandel As for StateFlow, it seems like a dangerous idea to rely on collection completion, cause StateFlow is conflated by design, so some emissions can be skipped by slow collectors.

[serandel]

In my use case (Kotlin gamedev, please don't judge :D ) I don't have a problem with slow collectors. In fact, I use the suspending emissions to synchronize game states and game events, so animations and sounds can drive the transitions between states in my state machines.

It works awesome and I really would like to have it built-in the flows instead of having my silly decorators, which make my types NOT proper Flows, so I lose the possibility of using the standard operators.

[allco]

1.6.3,
the problem is still there :-/
Perhaps is makes sense to reopen this issue?

[serandel]

I would love this to be reopened.

The issue about some people wanting to avoid waiting for slow collectors and some other people (me) just wanting to do exactly that, for synchronization, sounds to me like a possible option to specify when creating the SharedFlow.

[dkhalanskyjb]

I don't think this being an option on a SharedFlow is a good idea. It is stated everywhere that SharedFlow is, semantically, a hot stream of values, that is, it executes independently of the behavior of its subscribers. If we added an option to change that, this would go against the whole mental model of SharedFlow. Likewise, with StateFlow, which is also a hot flow, doing flow.value = x behaves in a fire-and-forget manner. What you're asking for doesn't fit the model of SharedFlow at all, in my opinion.

I think you should open a separate issue instead, for a Flow that broadcasts values to multiple subscribers and awaits their consumption before proceeding.

[serandel]

Well, if you look at the PR #2437, the documentation initially said that, as rendezvous, the emission was suspending until all collectors processed the event. So I don't think it's breaking the mental model, really, when even the team thought it worked like that at first. ;)

SharedFlow (and, in consequence, StateFlow) is hot in the sense (IMHO) that its generation of values is independent on how many collectors you have, instead of every collector executing a different instance of the emitting code in parallel. If it has a rendezvous behaviour or not doesn't affect this in my POV. Nevertheless, having a separate SharedFlow implementation instead of an option in the existing one, disabled by default, is perfectly fine as well.

Shall I open that ticket?

[dkhalanskyjb]

I think you are right. After all, we have https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-sharing-started/-companion/-while-subscribed.html, which stops the operation altogether if no subscribers are present, which is much more intrusive than just waiting for every subscriber to receive the value.