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

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

flatMapLatest

?

i don't want to cancel old flow though

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

it is a streaming response

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

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

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

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)

I think you want

flatMapConcat

?

I think that

flatMapConcat

/

flatMapMerge

does work fine -> it is just my test setup that is causing issues

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

flatMapMerge

.

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
      }
  }
}