How can be implemented such use-case with `Flow` <...

# coroutines

[U75957]

How can be implemented such use-case with Flow https://pl.kotl.in/nEwZYT3Kl (I have implementation with recursive launch, but I think it could be more elegant with Flow, but don't know how)
I want launch limited number of requests and as some request are completed, get the following from the queue.
For instance, we have queue/list of 100 URLs. But we want limit number of simultaneously executed requests by 5. Not 20 butches of 5 requests, but first butch with 5 request and next, when some request is ready. How can we process result of each request using collect?

[Dominaezzz]

I've implemented something like this but I want to confirm your use case.

[Dominaezzz]

You want to process the queue in batches and the next batch only gets processed after the first one finishes?

[Dominaezzz]

Or you want to process all elements in the queue in order but limit the number on in-flight (concurrent) processes?

[U75957]

@Dominaezzz I want to process all elements in the queue in order but limit the number on in-flight (concurrent) processes

[Dominaezzz]

Flow doesn't have this yet but here's something I made. https://github.com/Kotlin/kotlinx.coroutines/issues/1147#issuecomment-489153085

[U75957]

I expect that buffer with capacity parameter should do the job, but it don't work as expected (also w/o capacity parameter).
Just try example from official guide, but with other delays (I expect it should be done in about 110 seconds, but it takes >1000s):

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*

fun foo(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
//sampleStart
    val time = measureTimeMillis {
        foo()
            .buffer() // buffer emissions, don't wait
            .collect { value -> 
                delay(1) // pretend we are processing it for 1 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")
//sampleEnd
}

https://pl.kotl.in/qm9dV4CRv

[Dominaezzz]

That's because you delay(100) 10 times.

[Dominaezzz]

Flows are sequential.

[U75957]

@Dominaezzz check guide https://github.com/Kotlin/kotlinx.coroutines/blob/1.3.0/docs/flow.md#buffering

It produces the same numbers faster, as we have effectively created a processing pipeline, only having to wait 100 ms for the first number and then spending only 300 ms to process each number. This way it takes around 1000 ms to run

[U75957]

That's because you delay(100) 10 times.

With buffer() these 10 delays should be done concurrently.

[Dominaezzz]

buffer allows the previous operators to run concurrent to the next operators.

[Dominaezzz]

It doesn't make the flow concurrent.

[U75957]

@Dominaezzz but why following code (from official guide) takes around 1000ms?

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*

fun foo(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
//sampleStart
    val time = measureTimeMillis {
        foo()
            .buffer() // buffer emissions, don't wait
            .collect { value -> 
                delay(300) // pretend we are processing it for 300 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")
//sampleEnd
}

[Dominaezzz]

The first flow executes in 300ms. The second does so in 900ms + 100ms. They overlap because of the buffer. I'm not good at explaining this sorta thing.

[U75957]

@Dominaezzz Ah, ok, now I understand. Thank you.