我想通过websockets向客户发送通知 . 这些通知是由actor生成的，因此我试图在服务器启动时创建一个actor的消息流，并订阅websockects与此流的连接（仅发送自订阅以来发出的那些通知）
使用Source.actorRef，我们可以创建一个actor消息源 .
val ref = Source.actorRef[Weather](Int.MaxValue, fail)
.filter(!_.raining)
.to(Sink foreach println )
.run()
ref ! Weather("02139", 32.0, true)
但是，如果已经实现了，我如何订阅（akka http *）websockets连接到此源？
- Akka HTTP中的WebSockets连接需要Flow [Message，Message，Any]
我想要做的就像是
// at server startup
val notifications: Source[Notification,ActorRef] = Source.actorRef[Notificacion](5,OverflowStrategy.fail)
val ref = notifications.to(Sink.foreach(println(_))).run()
val notificationActor = system.actorOf(NotificationActor.props(ref))
// on ws connection
val notificationsWS = path("notificationsWS") {
parameter('name) { name ⇒
get {
onComplete(flow(name)){
case Success(f) => handleWebSocketMessages(f)
case Failure(e) => throw e
}
}
}
}
def flow(name: String) = {
val messages = notifications filter { n => n.name equals name } map { n => TextMessage.Strict(n.data) }
Flow.fromSinkAndSource(Sink.ignore, notifications)
}
这样做不起作用，因为通知源不是物化的，因此它不会发出任何元素 .
注意：我使用的是Source.actorPublisher并且它正在工作，但ktoso discourages his usage并且我收到此错误：
java.lang.IllegalStateException: onNext is not allowed when the stream has not requested elements, totalDemand was 0.
1 回答
您可以使用
mapMaterializedValue将物化的actorRef暴露给某个外部路由器actor .路由器可以跟踪你的来源actorrefs（deathwatch可以帮助整理东西）并将消息转发给他们 .
注意：您可能已经注意到了，但请注意，通过使用
Source.actorRef来提供流量，您的流量将无法识别背压（使用您选择的策略只会在负载下崩溃） .