如何使 spring webflux websocket 与 kotlin 协程一起工作？

Question

如何使 spring webflux websocket 与 kotlin 协程一起工作？

我有以下代码片段：

val client = ReactorNettyWebSocketClient()
val uri = URI.create("wss://websocket")

client.execute(uri) { session ->
    val handler = WebSocketImplementation(session)
    session.receive()
        .doOnNext { message ->
                handler.handleTextMessage(message) //this is a suspend function
        }
        .then()
}.block()

handleTextMessage 调用的一些方法是协程。虽然我可以使用 Reactor 重写它们，但我非常喜欢协程。有没有办法可以无缝地实现这种过渡？我已经在代码的其他部分做了，例如：

session.send(Mono.just(textMessage)).awaitFirstOrNull()

但无法使上面的代码片段与它一起工作

Answer 1

您可以将通量转换为流量，s.t. flow.onEach() 调用将需要暂停乐趣。

session.receive().asFlow().onEach {
        suspend { println(it) }
    }