Hi all, i am trying to stream redis pub sub messages to a websocket connection with this code

webSocket("/receive/{channel}") {
    val messageFlow: Flow<String> = callbackFlow {
    val jedis = jedisPool.resource

    jedis.subscribe(object : JedisPubSub() {
        override fun onMessage(channel: String, message: String) {
            trySendBlocking(message)
        }
    }, channel)

    awaitClose {
        jedis.close()
    }
}


messageFlow.cancellable().collect { message ->
    send(message)
}
 print("Websocket closed")
}

The issue is when the client closes the socket i never see the Websocket closed message, I presume the messageFlow isnt being cancelled? Is there a solution to this?