Is there a way to collect a flow inside the `webSo...

# ktor

[Shubham Singh]

Is there a way to collect a flow inside the webSocket block in Ktor Server and sendSerialized the values collected from that flow? In my case, I am observing the changes inside my PostgreSQL DB and returning the changes as a flow from the repository. Now I want to send this flow of values through a websocket connection so that whenever the value inside the flow changes, my websocket also sends a new frame, but right now when I try to do the following:

webSocket(getObservable) {
                    safeCall {
                        val id = call.parameters["id"]
                        repository.getObservable(id = UUID(id)).collect { value ->
                            sendSerialized(value)
                        }
                    }
                }

the websocket connection closes as soon as I connect to it.

[Shubham Singh]

What does this👂mean @Jonah Stiennon @Andromadus Naruto?

[Andromadus Naruto]

In my case, it means that I would love to hear how the conversation is going. Maybe I should have just used the 'watch for replies' feature.

[Shubham Singh]

Understood.. thanks for explaining 🙂

[Aleksei Tirman [JB]]

The problem may be that the safeCall returns immediately. Here is an example where a mutable state flow is collected inside a websocket handler:

val flow = MutableStateFlow(0)

GlobalScope.launch {
    while (true) {
        delay(2000)
        flow.update {
            it + 1
        }
    }
}

embeddedServer(Netty, port = 8082) {
    install(WebSockets)

    routing {
        webSocket("/ws") {
            flow.collect {
                send(Frame.Text(it.toString()))
            }
        }
    }
}.start(wait = true)

[Jonah Stiennon]

@Shubham Singh I'm testing an AI Q&A bot that I installed in this server. When I react with that emoji, the bot tries to answer the question, but sends the answer to me in a DM so that it doesn't pollute the channel with wrong answers. If the answer is right, I post it back to the thread, but in this case it was unhelpful 😞

[Shubham Singh]

@Aleksei Tirman [JB] safeCall wasn't the issue, it is just a try-catch block internally. The thing was that I was getting only a Flow from the repository (that is cold by nature), when I converted it into a SharedFlow using the shareIn convenience function, it doesn't close immediately anymore, but now I'm facing another issue i.e. maybe the Apollo graphql client is not able to parse a UUID string as Json and hence throws the error in the image attached below. I'll maybe try to post this in the Apollo GraphQL slack channel.

[Shubham Singh]

Still, thank you for taking out time to help ⭐ The only thing I wanted to know from you if there is official documentation or an example that teaches us how to work with Flows inside a WebSocket in Ktor? Or maybe a WebSocket isn't even meant to perform such operations and there's something else that helps.