Is there a way to collect each item in a separate ...

# coroutines

[Dean Djermanović]

Is there a way to collect each item in a separate coroutine in Kotlin Flow? For example, if I have this source:

val source = flow {
        emit(1)
        delay(200)
        emit(2)
        delay(200)
        emit(3)
    }

source
    .onEach { delay(500) }
    .collect {
        ...
    }

I want my items to be collected after:

1 - 500ms
2 - 700ms
3 - 900ms

[Joseph Hawkes-Cates]

I’m not sure if this is what you are looking for but the following should get the timing that you list in your example:

source
     .onStart{ delay(500) }
     .collect {
         ...
     }

[Nick Allen]

If you read the Flow interface docs, it guarantees that items are collected sequentially, so it's not possible in the way you are thinking.

However you can launch from collect

coroutineScope {
    source.collect {
        launch {
            delay(500)
            //Move original collect code here
        }
    }

[Nick Allen]

You could potentially leverage something like flattenMerge which has optional param to limit concurrency. One Flow must emit items sequentially but if you turn it into many `Flow`s then the restriction doesn't apply.

source.flattenMerge {
    flow {
        delay(500)
        //Move original collect code here
        //Or could emit here so the collect below gets items and that part is sequential.
    }
}
    .collect()

[Dean Djermanović]

Thanks @Nick Allen