目前我正在试图建立一个阿卡-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
回调从来没有触发(我从来没有得到一个“终止:......”我的控制台上的消息)。这是为什么?而且怎么可能当客户端离开(这样我就可以从我的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!")
}