CODEWITHSUNDEEP
scalaakkaakka-streamakka-http
Florian Baierl
Florian Baierl

Reputation: 2481

Simple server-push broadcast flow with Akka

I am struggling to implement a - rather simple - Akka flow. Here is what I think I need:

akka-flow

I have a single server and n clients and want to be able to react to external events by broadcasting messages (JSON) to the clients. The clients can register/deregister at any time.

So for example:

Here is what I have so far:

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))
}

Upvotes: 4

Views: 615

Answers (1)

Ivan Stanislavciuc
Ivan Stanislavciuc

Reputation: 7275

To be able to use a broadcastHub, you have to define two flows. One that runs your websocket TextMessage to the broadcastHub. You have to run it, it produces a Source that you connect to each client.

Here it is this concept described in simple runnable app. 

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()
}

Prints

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

If clients are known in advance, you don't need BrodacastHub and you can use alsoTo method:

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

Upvotes: 3

Related Questions