代码之家 › 专栏 › 技术社区 › kordax

拦截scala中socket close上的akka http websocket事件

akka-http akka-stream akka websocket scala

kordax · 技术社区 · 6 年前

我正在使用带有akka http的scala。

我有一个基于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
  ...
}

我的websocket客户通过keep alive建立双向通信连接。

绑定使用:

val binding = Http().bindAndHandle(webSocketRoute, config.host, config.port)

问题是,我需要为一个封闭的套接字注入一个回调(例如,如果一个客户机断开),并且减少当前的客户数,但是我找不到任何入口点。

是否可以在套接字关闭时捕获某种事件?

1 回复 | 直到 6 年前

Jeffrey Chung 6 年前

使用 watchTermination :

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] = ???

推荐文章

missingfaktor Kevin Wright · 非致命性渔获物可以丢弃吗?

6 年前

joesan · Akka有状态参与者跨节点复制

6 年前

sekhar · 如何解决此异常加载类文件“SprayJsonSupport”时检测到的无效依赖关系。类别'

6 年前

Joshua Esolk · 为什么在Akka Dispatcher上启动Futures时,Futures中的Futures会按顺序运行

6 年前

bumblebee · 分布式发布订阅Akka集群中的发布和发送策略

6 年前

Eric Kolotyluk · Akka HTTP与Akka类型一起工作吗?

6 年前

user_mda · 在有限状态机中更新数据

6 年前

Falk Schuetzenmeister · Akka:如何在一个图形阶段提取值并在下一个图形阶段使用

6 年前

Christos Hadjinikolis · Flink JobExecutionException:akka。客户超时

6 年前

Teyu11 · Akka HTTP多文件上传到S3

6 年前