kor*_*dax 2 scala websocket akka akka-stream akka-http
我将 Scala 与 Akka HTTP 一起使用。
我有一个基于 Akka HTTP 的服务器后端应用程序。传入的 WebSocket 消息通过以下方式处理handleWebSocketMessages:
/**
* Route for WebSocket request
*/
private val webSocketRoute: Route = pathSuffix(Constants.WSPROXY_WEBSOCKET_PATH_SUFFIX) {
LOGGER.debug("() Web socket route")
handleWebSocketMessages(wsRouteImpl)
}
/**
* This method calls all registered handlers.
*
* @return flow for handleWebSocketMessages method
*/
private def wsRouteImpl: Flow[Message, Message, Any] = {
LOGGER.debug("() wsRouteHandling")
numOfClients += 1
LOGGER.info(s"Client has connected, current number of clients: $numOfClients")
var flow = Flow[Message].mapConcat {
// Call specific handlers depending on message type
...
}
Run Code Online (Sandbox Code Playgroud)
我的 WebSocket 客户端通过 keep-alive 建立双向通信连接。
绑定是通过以下方式完成的:
val binding = Http().bindAndHandle(webSocketRoute, config.host, config.port)
Run Code Online (Sandbox Code Playgroud)
问题是我需要为关闭的套接字注入回调(例如,如果客户端已断开连接)并减少当前的客户端数量,但我找不到任何入口点。
是否可以在套接字关闭时捕获某种事件?
val numOfClients = new java.util.concurrent.atomic.AtomicInteger(0)
private val webSocketRoute: Route = pathSuffix(Constants.WSPROXY_WEBSOCKET_PATH_SUFFIX) {
LOGGER.debug("() Web socket route")
val wsFlow: Flow[Message, Message, Any] =
wsRouteImpl.watchTermination() { (_, fut) =>
numOfClients.incrementAndGet()
LOGGER.info(s"Client has connected. Current number of clients: $numOfClients")
fut onComplete {
case Success(_) =>
numOfClients.decrementAndGet()
LOGGER.info(s"Client has disconnected. Current number of clients: $numOfClients")
case Failure(ex) =>
numOfClients.decrementAndGet()
LOGGER.error(s"Disconnection failure (number of clients: $numOfClients): $ex")
}
}
handleWebSocketMessages(wsFlow)
}
private def wsRouteImpl: Flow[Message, Message, Any] = ???
Run Code Online (Sandbox Code Playgroud)
| 归档时间: |
|
| 查看次数: |
926 次 |
| 最近记录: |