Would this onEach work to keep the current state of the flow?

private val state = MutableStateFlow(initialState)

   fun getState(events: Flow<Event>): Flow<State> =
        merge(
            state, 
            events.transformLatest {
                emit(state.value.copy(...))
                ...
                emit(state.value.copy(...))
            }
        ).onEach { state.value = it }

Think I'm missing something here, the second emit calling

state.value

gets the same value as the first one

merge

is a concurrent operation if I recall correctly, so the `emit`s are buffered by default

Oh, I see. So just to make sure I got this right: 1.

state

and

events.transform

will run in two separate coroutines here, this is not a sequential flow

All flows are merged concurrently

2. emit suspends until the value is emitted, not until it is collected downstream, that's why I get a race condition by expecting my state (

state.value

) to have been updated already I don't get, however, what this statement in the docs means:

Operator fusion

Applications of flowOn, buffer, produceIn, and broadcastIn after this operator are fused with its concurrent merging so that only one properly configured channel is used for execution of merging logic.

1 & 2 are correct operator fusion means that if you put a

.buffer(RENDEZVOUS)

after

merge

, it knows that

merge

uses a channel and will change merge's channel to be a RENDEZVOUS channel, rather than introduce a second channel

Got it. Thank you very much! :)

@octylFractal When used with

merge

, is

RENDEZVOUS

synchronizing between merged upstreams on the one hand and downstream collection on the other? Or does it also affect how upstreams are merged together?

it might affect how they merge, I'm not sure of ordering guarantees for `offer`/`send` on those types of channels

Got it. Thanks @octylFractal