使用Akka的简单服务器推送广播流程

Question

我正在努力实现一个 - 相当简单 - Akka流程。这是我认为我需要的：

我有一个服务器和n个客户端，希望能够通过向客户端广播消息（JSON）来对外部事件做出反应。客户可以随时注册/注销。

例如：

• 1个客户已注册
• 服务器抛出一个事件（“Hello World！”）
• 服务器播放“Hello World！”给所有客户（一个客户）
• 新客户端打开websocket连接
• 服务器抛出另一个事件（“Hello Akka！”）
• 服务器播放“Hello Akka！”给所有客户（两个客户）

这是我到目前为止：

def route: Route = {
   val register = path("register") {
     // registration point for the clients
     handleWebSocketMessages(serverPushFlow)
   }
}

// ...

def broadcast(msg: String): Unit = {
  // use the previously created flow to send messages to all clients
  // ???
}

// my broadcast sink to send messages to the clients
val broadcastSink: Sink[String, Source[String, NotUsed]] = BroadcastHub.sink[String]

// a source that emmits simple strings
val simpleMsgSource = Source(Nil: List[String])

def serverPushFlow = {
  Flow[Message].mapAsync(1) {
    case TextMessage.Strict(text) =>       Future.successful(text)
    case streamed: TextMessage.Streamed => streamed.textStream.runFold("")(_ ++ _)
  }
  .via(Flow.fromSinkAndSource(broadcastSink, simpleMsgSource))
  .map[Message](string => TextMessage(string))
}

Answer 1

为了能够使用broadcastHub，您必须定义两个流。一个运行你的websocket TextMessage到broadcastHub。您必须运行它，它会生成一个连接到每个客户端的Source。

这是在简单的可运行应用程序中描述的这个概念。

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{BroadcastHub, Sink, Source}
import org.slf4j.LoggerFactory

import scala.concurrent.duration._

object BroadcastSink extends App {

  private val logger = LoggerFactory.getLogger("logger")

  implicit val actorSystem = ActorSystem()
  implicit val actorMaterializer = ActorMaterializer()

  val broadcastSink: Sink[String, Source[String, NotUsed]] =
    BroadcastHub.sink[String]

  val simpleMsgSource = Source.tick(500.milli, 500.milli, "Single Message")

  val sourceForClients: Source[String, NotUsed] = simpleMsgSource.runWith(broadcastSink)

  sourceForClients.to(Sink.foreach(t => logger.info(s"Client 1: $t"))).run()
  Thread.sleep(1000)

  sourceForClients.to(Sink.foreach(t => logger.info(s"Client 2: $t"))).run()
  Thread.sleep(1000)

  sourceForClients.to(Sink.foreach(t => logger.info(s"Client 3: $t"))).run()
  Thread.sleep(1000)

  sourceForClients.to(Sink.foreach(t => logger.info(s"Client 4: $t"))).run()
  Thread.sleep(1000)

  actorSystem.terminate()
}

打印

10:52:01.774 Client 1: Single Message
10:52:02.273 Client 1: Single Message
10:52:02.273 Client 2: Single Message
10:52:02.773 Client 2: Single Message
10:52:02.773 Client 1: Single Message
10:52:03.272 Client 3: Single Message
10:52:03.272 Client 2: Single Message
10:52:03.272 Client 1: Single Message
10:52:03.772 Client 1: Single Message
10:52:03.772 Client 3: Single Message
10:52:03.773 Client 2: Single Message
10:52:04.272 Client 2: Single Message
10:52:04.272 Client 4: Single Message
10:52:04.272 Client 1: Single Message
10:52:04.273 Client 3: Single Message
10:52:04.772 Client 1: Single Message
10:52:04.772 Client 2: Single Message
10:52:04.772 Client 3: Single Message
10:52:04.772 Client 4: Single Message
10:52:05.271 Client 4: Single Message
10:52:05.271 Client 1: Single Message
10:52:05.271 Client 3: Single Message
10:52:05.272 Client 2: Single Message

如果客户事先知道，您不需要BrodacastHub，您可以使用alsoTo方法：

  def webSocketHandler(clients: List[Sink[Message, NotUsed]]): Flow[Message, Message, Any] = {
    val flow = Flow[Message]
    clients.foldLeft(flow) {case (fl, client) =>
      fl.alsoTo(client)
    }
  }