Cancellation propagation between spatially disconnected FlowCollectors

[akarnokd]

Given an operator where the upstream calling emit and the downstream calling emit are in some way disconnected. One such example would be an operator that multicasts values via a help of a function transformation:

source.multicast { upstream ->
   merge(upstream.filter { it % 2 == 0 }, upstream.filter { it % 2 != 0 })
}

However, there is no guarantee that the upstream is connected to the downstream consumer, so for example,

source.multicast { upstream ->
    flow { emit(0) }
}

would terminate the downstream but the upstream has to know somehow it should stop.

Since cancellation in Flow is not a signal but emit() throwing, I'm not certain how could the upstream be notified in a reasonable manner in a reasonable amount of time; i.e., the upstream may not call its emit for some time to discover it has to cancel.

Are there any tips or practices to handle such cases?

[elizarov]

Please, take a look at a channelFlow: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/channel-flow.html
It seems to precisely fit your description of "disconnected" use-case. This function explicitly converts a flow to a channel, which can be "signalled" in a totally independent way and also provides all the necessarily synchronization and context-bridging:

channelFlow { 
    // can call `send(value)` at any time to signal value
    // can call `close(exception)` at any time to signal completion/error
}

[qwwdfsad]

Feel free to reopen if you have further questions