Scala akka-http WebSocket : How to save the client connection and push message to the client when needed?

Question

如何将客户端(网络)连接保持在内存变量中，然后在需要时将传出消息发送到客户端(网络)？

一旦服务器收到来自客户端的消息，我已经有一些简单的代码用于将消息推回客户端。如何为传出消息部分修改下面的代码？

implicit val actorSystem = ActorSystem("akka-system")
implicit val flowMaterializer = ActorMaterializer()
implicit val executionContext = actorSystem.dispatcher

val ip = "127.0.0.1"
val port = 32000

val route = get {
    pathEndOrSingleSlash {
        complete("Welcome to websocket server")
    }
} ~
path("hello") {
    get {
        handleWebSocketMessages(echoService)
    }
}

def sendMessageToClient(msg : String) {

    // *** How to implement this?
    // *** How to save the client connection when it is first connected?
    //     Then how to send message to this connection?

}

val echoService = Flow[Message].collect {

    // *** Here the server push back messages when receiving msg from client

    case tm : TextMessage => TextMessage(Source.single("Hello ") ++ tm.textStream)
    case _ => TextMessage("Message type unsupported")
}

val binding = Http().bindAndHandle(route, ip, port)

Answer 1

您可以通过 .map 研究管道汇流。称呼。内.map调用您可以捕获该值，然后返回相同的消息。例如:

  Flow[Message].collect {
    case tm : TextMessage =>
      TextMessage(Source.single("Hello ") ++ tm.textStream.via(
        Flow[String].map((message) => {println(message) /* capture value here*/; message})))
    case _ => TextMessage("Message type unsupported")
  }

现在，如果您打算处理这些值并稍后发送值，您想要的不是单个源到接收器流，而是接收器和源的两个独立流，您可以使用 Flow.fromSinkAndSource例如

Flow.fromSinkAndSource[Message, Message](
  Flow[Message].collect { /* capture values */},
    // Or send stream to other sink for more processing
  source
)

很有可能，这个源要么是由图形 DSL 构建的，一个手工制作的 Actor ，要么你可以考虑使用可重用的助手，比如 MergeHub .