I guess I want to merge different flow that will be called from

node.run()

and collect all the results to

running

keeping the order

I found a solution like this:

private var _running = MutableSharedFlow<Pair<Event, String?>>()
    val running: SharedFlow<Pair<Event, String?>> = _running

    fun execute(scope: CoroutineScope, node: NodeInterface) {
        scope.launch {
            node.run().collect {
                _running.emit(it)
            }
        }
    }

Please, tell me if there is a better way 🙂

Your second approach has a flaw, where the lifetime of the

collect

call in execute is not tied at all to the actual consumer's

collect

call (the one collecting from

running

). So it runs even when nothing is listening, and you are not just getting items from the latest

node

, you are getting all items from all nodes because nothing cancels the previous job when

execute

is called a second time. Fortunately, all this is handled by

flatMapLatest

which does exactly what you are asking for.

private var latestNode = MutableSharedFlow<NodeInterface>(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
    val running: Flow<Pair<Event, String?>> = latestNode.flatMapLatest { it.run() }

    fun execute(node: NodeInterface) {
        latestNode.tryEmit(node)
    }

On each new item, it'll cancel what it was doing, call the lambda to get a new

Flow

, and emit items from it.