brendanw
commented
4 years ago I believe the mistake is on my end. The assumption I was making is thinking that each flow combined by the merge function would immediately dispatch emissions to the channel backing the merged flow. If I look at the naive merge implementation that is commented out in source (without fusing) then I can see that this assumption is wrong. A coroutine is launched per flow passed to the merge function. We have no guarantee where each of those coroutines will be in its execution at the moment that the collect call is made on the channelFlow representing the merged flows.
This ended up being my approach:
fun <T> List<Flow<T>>.onBindMerge(onBind: () -> Unit): Flow<T> {
var boundFlows = atomic(0)
return channelFlow {
forEach { flow ->
launch {
flow.asBindFlow {
if (boundFlows.incrementAndGet() == size) onBind()
}.collect {
send(it)
}
}
}
}
}
@InternalCoroutinesApi
fun <T> Flow<T>.asBindFlow(onBind: () -> Unit): Flow<T> {
return BindFlow(onBind = onBind, flow = this)
}
@InternalCoroutinesApi
class BindFlow<T>(val onBind: () -> Unit, val flow: Flow<T>) : Flow<T> {
private var hasBinded = atomic(false)
@InternalCoroutinesApi
override suspend fun collect(collector: FlowCollector<T>) {
if (hasBinded.compareAndSet(expect = false, update = true)) {
onBind()
}
flow.collect {
collector.emit(it)
}
}
}I use this for implementing a MVI/redux-like StateMachine
class StateMachine<R : Any, T>(
val scope: CoroutineScope,
private val initialState: T,
private val sideEffects: List<(Flow<R>, () -> T) -> Flow<R>>,
private val reducer: suspend (accumulator: T, value: R) -> T
) {
val viewState: FlowRelay<T> = FlowRelay()
private var isInitialized = atomic(false)
private val inputActions: BroadcastChannel<R> = BroadcastChannel(Channel.BUFFERED)
init {
var job: Job? = scope.launch {
val lastState = StateWrapper(initialState)
val flowList = sideEffects.map { sideEffect ->
sideEffect(inputActions.asFlow(), { lastState.state })
}.run {
toMutableList().apply {
add(inputActions.asFlow())
}
}
flowList.onBindMerge {
isInitialized.value = true
}
.onEach { kprint("result $it") }
.onCompletion { inputActions.cancel() }
.scan(lastState.state, reducer)
.collect { outputState ->
lastState.state = outputState
viewState.send(outputState)
}
}
}
fun dispatchAction(action: R) = scope.launch {
kprint("Received input action $action")
while (!isInitialized.value) {
yield()
}
inputActions.send(action)
}
}Sharing to give the context that original motivation for all of this was ensuring that no action sent to dispatchAction is sent to the channel until the final collector in the stream is guaranteed to be collecting (and now I realize I want the guarantee that all collectors in the stream are guaranteed to be collecting before the first action is sent to the inputActions channel).
Please let me know if there is a more idiomatic way to accomplish this.
elizarov
qwwdfsad
Branch: native-mt (last commit: 10979a350426f23c02356c66acfda1e765d23450) gradle task: macosX64Test
This test passes
This test fails
Both tests depend on
asBindFlowdefined here