针对特定的 Micronaut WebSocket 连接

Question

我正在尝试创建一个 Kotlin 服务，该服务能够将数据从 Kafka 传递到（特定的）WebSocket 连接。例如，如果来自用户的数据在 Kafka 上传递，程序会接收到它，如果当前连接的是同一个用户，我想将它传递到正确的 WebSocket 连接。

我的基础知识是：

@KafkaListener()
@Controller
@Secured(SecurityRule.IS_ANONYMOUS)
@ServerWebSocket("/ws/{id}")
class WebSocket() {

    @OnOpen
    fun onOpen(session: WebSocketSession, id: String): Publisher<String> {
        return session.send("connection opened")
    }
}

但是，如果在同一个 class 中我收到一条 Kafka 消息，其中包含我要查找的用户 ID，我如何将其传递给正确的 WebSocket 连接？

我想也许如果我使用 url 中的 "/ws/{id}" 我可以轻松地将它发送到正确的 ID，但我不知道如何。

我知道将数据发送到连接的唯一方法是从 websocket 注释函数内部，如 OnOpen 和 OnMessage 等。此外，我发现 Micronaut WebSocketBroadcaster 很有用，但当然只用于广播。

我想弄清楚的是，在哪里可以确保连接到的 ID 也是用户的实际 ID，而不是其他人的，我应该在 @Secured 注释中实现吗？

Answer 1

对于任何感兴趣的人，我确实找到了解决方案。使用 Micronaut WebSocketBroadcaster 您可以广播消息，并通过使用第二个参数，它采用 Predicate<WebSocketSession> 定义它向哪个 WebSocketSession 发送消息。

如果您确实知道特定的 WebSocketSession ID，则可以简单地使用 { it.id == "id" } 作为谓词，如示例 1 中所做的那样。如果您确实需要根据其他一些属性将其发送给特定用户，则可以通过使用 session.attributes.put("id", id) 将属性分配给 WebSocketSession 本身，例如在 onOpen 函数中。您可以为此使用的谓词在示例 2 中，唯一的问题是我无法找到一种方法来检索属性而不将其转换为地图，这不是很漂亮。

class ExampleClass(val broadcaster: WebSocketBroadcaster) {
    val userId = "user_001"
    val webSocketId = "asdasdasd"

    // [1] Send message to a specific socket ID
    broadcaster.broadcast("message", { it.id == webSocketId)

    // [2] Send message to a user with specific ID or any other identifying attribute of choice
    broadcaster.broadcast("message", { it.attributes.asMap()["id"]?.equals(id) ?: false})    
}

针对特定的 Micronaut WebSocket 连接

Targeting specific Micronaut WebSocket connections

java

websocket

kotlin

micronaut