I'm looking for examples of how to combine Ktor WebSocket client with Asynchronous Flow. I have a pu...

# ktor

[eirikb]

I'm looking for examples of how to combine Ktor WebSocket client with Asynchronous Flow.
I have a pub/sub-like setup where I send a message to server in order to subscribe for values, but I'm completely lost on how to set this up. I can share what I have so far, but it's quite bad. I doesn't have to be with Flow, but I figured it would fit quite well here.

[cy]

Are you talking about coroutines flow?

[cy]

You can simply make a flow from the incoming frames channel

[eirikb]

Yes coroutines flow. How?

[cy]

client.wss("....") {
    incoming.consumeAsFlow() ...
}

?

[eirikb]

Nice. How can I return it? My code is a library where others can subscribe (and get a flow back)

[cy]

Well, it depends

[cy]

If you need it to be cold then you need something like this:

fun HttpClient.webSocketFlow(urlString: String): Flow<Frame> = flow {
        ws(urlString) {
            emitAll(incoming)
        }
    }

[cy]

If you need a hot flow, then

client.webSocketSession(host = "xxx", path = "/endpoint").incoming.consumeAsFlow()

[cy]

The first is cool, you may also combine with flow retry functions to reconnect

[eirikb]

Thank you. I will have to experiment with this

[eirikb]

I've been testing out some approaches, but wasn't able to use consumeAsFlow because I want multiple listeners. I have managed to create working solution, but it looks messy: https://gist.github.com/eirikb/41910638c2acef9f8f84c43f507f6023 . It also has some issues

[eirikb]

Perhaps this is something I could use https://github.com/dropbox/Store/blob/master/multicast/src/main/kotlin/com/dropbox/flow/multicast/Multicaster.kt ?