hey folks! I have a list of parallel computations. How would I write a function like this? Can’t rea...

# coroutines

[adeln]

hey folks! I have a list of parallel computations. How would I write a function like this? Can’t really wrap my head around it

fun <T> List<Deferred<T>>.toFlow(): Flow<T>

I want the flow to emit values as soon as they become available

[uli]

do you want to process in order and/or preserve order?

[adeln]

I want the flow to emit values as soon as they become available

[marstran]

It's easy if you await them in the order of the list:

fun <T> List<Deferred<T>>.toFlow(): Flow<T> = flow {
    forEach { emit(it.await()) }
}

Emitting items in order of completion is harder. You need to keep track of the deferreds that have already completed, so that you can complete the flow at the right time.

[gildor]

You can use select

[gildor]

@marstranThis is not “I want the flow to emit values as soon as they become availab”

[marstran]

Yes, I know. That's what I wrote after the example.

[uli]

Maybe flatMapMerge can help? But select might be more efficient depending on the size of the List

[uli]

See: https://github.com/Kotlin/kotlinx.coroutines/issues/1147

[gildor]

Just use select to get values whey they are ready

[marstran]

Maybe this (after @uli's suggestion):

fun <T> List<Deferred<T>>.toFlow(): Flow<T> =  
    asFlow()
      .flatMapMerge { flow { emit(it.await()) } }

[adeln]

it works, thanks! I’m wondering if there are more efficient ways of doing this

[adeln]

@gildor yeah, I can’t really understand how to use select to skip already emitted values. It just keeps emitting the same value

[gildor]

@adeln

fun <T> List<Deferred<T>>.toFlow(): Flow<T> = flow {
    forEach { deferred ->
        select {
            deferred.onAwait { value ->
                emit(value)
            }
        }
    }
}

[adeln]

it works, thank you so much! Didn’t know about whileSelect

[gildor]

Actually in this case you don’t need whileSelect, just select is enough
I updated sample

[gildor]

actually it’s good candidate for kotlinx.coroutines, probably make sense to report a feature request

[adeln]

it works! 🙂 thank you

[adeln]

I don’t know about kotlinx.coroutines. Since List<Deferred> is already hot, toFlow() doesn’t make much sense. Maybe I needed toChannel(): ReceiveChannel or maybe I need to reconsider the way I launch my parallel computations

[gildor]

No, I don’t think so

[gildor]

Flow is fine even for hot sources

[gildor]

With channel you can of course cancel all those deferred, but I wouldn’t explose it as API

[gildor]

what is your use case there?

[adeln]

@gildor https://paste2.org/AAHkGa0k

[gildor]

I don’t see why do you need flow here
Why not just call webSocket.send(it) after hitNetwork

[adeln]

with flow I can debug it by just printing to console

flow is also easier to test than faking the websocket

[adeln]

I guess I’m looking for a way to produce the flow from parallel computations. But I can’t just call launch {} inside flow {} =/

[gildor]

But I can’t just call launch {} inside flow {}

You can, use channelFlow

[gildor]

with flow I can debug it by just printing to console

Why you cannot do this with:

val result = hitNetwork()
println("Result!!! $result")
webSocket.send(result)
println("finished")

[gildor]

But I really don’t see why you need flow in this sample

[gildor]

It may be required if you would have not input: List<Input> but input: Flow<Input>, but if you already have input and all processing happeing on the same function, why would you need flow

[gildor]

if you have list you even can do this:

input // List<Input>
            .asFlow()
            .map {
                async { hitNetwork(it) }
            }
            .map {
                webSocket.send(it.await())
            }
            .collect()

[gildor]

But honestly I see no reason for this, because I can just do:

coroutineScope {
 input.forEach {
    launch {
        val result = hitNetwork(it)
        webSocket.send(result)
    }
  }
}

[adeln]

thanks! channelFlow is exactly what I needed

[adeln]

@gildor In debug I want to filter between success/error, but in prod I want all the values. Flow is just easier to manipulate

[gildor]

In debug I want to filter between success/error

Why not just use if for that, why do you need flow?

[gildor]

I mean it may be needed if you return it, now you just do some work and return

[adeln]

I completely understand that in my use case the producer and the consumer may live in the same function. But I want to separate them

[gildor]

than probably you need something completely different, but hard to say without sample

[adeln]

flow is just easier to play with: filter doesn’t created another level of nesting, and is easy to comment out

same with take(n), easy to stop the execution of the program when you’re just playing with it

[gildor]

¯\_(ツ)_/¯

[adeln]

I completely agree that I wouldn’t need Flow when I have completely figured out all the small little details of my problem