scala - 在 akka-http 流中未触发 watchTermination
问题描述
我目前正在尝试构建一个 akka-http websocket 连接,它可以:
- 广播到所有连接的客户端
- 回答客户的某些要求
到目前为止,这就是我创建流程的方式:
// keeps a list of all actors so I can broadcast to them
var actors: List[ActorRef] = Nil
private def wsFlow(implicit materializer: ActorMaterializer): Flow[ws.Message, ws.Message, NotUsed] = {
val (actor, source) = Source.actorRef[String](10, akka.stream.OverflowStrategy.dropTail)
.toMat(BroadcastHub.sink[String])(Keep.both)
.run()
// this never triggers
source.watchTermination() { (m, f) =>
f.onComplete(r => println("TERMINATION: " + r.toString))
actors = actors diff actor :: Nil
m
}
actors = actor :: actors
val wsHandler: Flow[ws.Message, ws.Message, NotUsed] =
Flow[ws.Message]
.merge(source)
.map {
case TextMessage.Strict(tm) => handleMessage(actor, tm)
case _ => TextMessage.Strict("Ignored message!")
}
wsHandler
}
def broadcast(msg: String): Unit = {
actors.foreach(_ ! TextMessage.Strict(msg))
}
我遇到的 - 希望 - 最后一个问题是watchTermination
回调永远不会触发(我的控制台上永远不会收到“TERMINATION:...”消息)。这是为什么?以及如何检测客户何时离开(以便我可以将他从我的actors
列表中删除)?
解决方案
我想出了如何做到这一点:
val wsHandler: Flow[ws.Message, ws.Message, NotUsed] = Flow[ws.Message]
.watchTermination() { (m, f) =>
f.onComplete(r => {
println("Client left: " + r.toString)
actors = actors diff actor :: Nil
}
)
m
}
.merge(source)
.map {
case TextMessage.Strict(tm) => handleMessage(actor, tm)
case _ => TextMessage.Strict("Ignored message!")
}
推荐阅读
- jquery - jQuery隐藏/显示功能覆盖html内联样式背景图像?
- mysql - SQL 中 AND 附近的语法不正确
- java - HashMap 类中的最终方法
- javascript - jQuery 自定义内容滚动条速度
- lua - Roblox Studio - 当我尝试出售背包中的“雪”时出现警告/错误
- caching - 管理/清理多个缓存
- android - 创建 TeamSite 混合应用程序 16
- angular - Angular 6 通过 json 并根据 id 获取数据
- javascript - ASP.net 在客户端上下文中运行 javascript
- javascript - 如何选择调用ngIf的函数?