hello :wave: whats the proper way to start a flow ...

# coroutines

[Dariusz Kuc]

hello 👋
whats the proper way to start a flow based on another flow? i.e. currently I just map incoming flow to the result flow.... but that has a problem that it is sequentially processing elements instead of returning parallel flows

[kevin.cianfarini]

I don't know if I'm understanding your question correctly, but it seems like you might want flatMapLatest ?

[Dariusz Kuc]

i don't want to cancel old flow though

[Dariusz Kuc]

i.e. trying to implement a protocol on a websocket

[Dariusz Kuc]

so got a stream of incoming messages and for each one of them I need to return a flow of responses

[kevin.cianfarini]

Do you need to return a flow of responses, or just launch an associated job?

[Dariusz Kuc]

it is a streaming response

[kevin.cianfarini]

So you have a single incoming stream of element and a single outgoing stream, yeah?

[kevin.cianfarini]

Or does it fan out into multiple outbound connections?

[Dariusz Kuc]

well it should fan out but its all on the same session

[Dariusz Kuc]

generic spring websocket impl just follows a basic flux

session.send(session.receive().map { toWhatever })

[kevin.cianfarini]

I'm personally not familiar with how websockets work at all so if you can distill the problem down to one that doesn't require that domain knowledge I might be able to help

[kevin.cianfarini]

I'm still unsure what you need unfortunately

[kevin.cianfarini]

Like, are the flow of 'responses' intended to be infinite? Finite?

[Dariusz Kuc]

so at high level
• given incoming flow of messages (start, subscribe, ping, complete)
• return a flow of messages based on the incoming message (i.e. subscribe -> return flow of arbitrary length, others are just flow of 1 element or just cancel)
If I'm processing the response for subscribe I still need to respond to other incoming requests (e.g. users can ping me OR send complete to stop the existing flow)

[Dariusz Kuc]

actually now that I'm thinking about it

[kevin.cianfarini]

I think you want flatMapConcat?

[Dariusz Kuc]

I think that flatMapConcat / flatMapMerge does work fine -> it is just my test setup that is causing issues

[kevin.cianfarini]

It want things to all run concurrently, and not sequentially, you definitely want flatMapMerge.

[Dariusz Kuc]

well actually found a problem in my code, I'm doing something along the lines

val topLevelScope = CoroutineScope(SupervisorJob)
val runningStuff = ConcurrentHashMap<String, Job>

// within function
someFlow.flatMapMerge { 
  x -> channelFlow {
    when (x)
      is Foo -> send(response)
      is Bar -> {
        val longRunningProcess = launch {
          generateAnotherFlow().collect {
            send(it) // <---- problematic as channelFlow will be closed
          }
        }
        runningStuff[id] = longRunningProcess
      }
  }
}

[Dariusz Kuc]

previously I was wrapping the launch in a channelFlow scope.... but that had another side effect that scope was blocked until the launch completed so not the behavior I want

[Dariusz Kuc]

any thoughts?

[Dariusz Kuc]

doh 🤦 have to just launch it from the flow scope.... topLevelScope is just for another timeout job

[Dariusz Kuc]

looks like it all works now