1
\$\begingroup\$

Akka-http supports WS connections like that:

def route =
    path("register") {
      parameter('name) { name ⇒
        handleWebSocketMessages(pubSub.openNewConnection(UserId(id))))
      }
    }

But implementation of WS requires some machinery:

  • we should store all connected clients
  • we should be able send messages to these clients
  • in case of disconnect we should update this map.

So I suggest the following code for it. But I'm new to akka-http, if you have any ideas how to improve this code, you're welcome!

import akka.NotUsed
import akka.actor.{ActorRef, ActorSystem, Status}
import akka.stream.Materializer
import akka.stream.scaladsl.{Flow, Source}
import cats.Monad
import com.hello.UserId
import com.hello.Logging
import akka.http.scaladsl.model.ws.{Message, TextMessage}
import akka.stream.OverflowStrategy
import akka.stream.scaladsl._
import org.reactivestreams.Publisher
import akka.stream.scaladsl.Sink

import scala.concurrent.Future

class PubSub()(implicit system: ActorSystem, mat: Materializer) extends Logging {
  implicit val ec = system.dispatcher

  private var wsConnections: Map[UserId, ActorRef] = Map.empty[UserId, ActorRef]

  def getConnections = wsConnections.size

  def openNewConnection(id: UserId): Flow[Message, Message, NotUsed] = {
    closeConnection(id)

    val (actorRef: ActorRef, publisher: Publisher[TextMessage.Strict]) = 
      Source.actorRef[WsMsg.Out](16, OverflowStrategy.fail)
        .map(msg => TextMessage.Strict(Codec.encodeWsMsg(msg)))
        .toMat(Sink.asPublisher(false))(Keep.both).run()

    val sink: Sink[Message, Unit] = Flow[Message]
      .watchTermination()((_, ft) => ft.foreach { _ => closeConnection(id) })
      .mapConcat {
        case TextMessage.Strict(msg) =>
          Codec.decodeWsMsg(msg).fold(err => {
            log.info(s"Unknown message $msg. Parsing error: $err")
            Nil
          }, msg => List(msg))
        case _                       =>
          Nil
      }
      .mapAsync(4) {
        case WsMsg.In.Ping(n) =>
          actorRef ! WsMsg.Out.Pong(n)
          Future.successful("Pong!")
        case msg              =>
          //TODO send to dispatcher
          Future.successful {
            log.info(s"Dispatching $msg")
          }
      }
      .to(Sink.ignore)

    log.info(s"Opening WS connection for $id")
    wsConnections += id -> actorRef
    Flow.fromSinkAndSource(sink, Source.fromPublisher(publisher))
  }

  def sendToPlayer(id: UserId, msg: WsMsg.Out) = wsConnections.get(id).foreach(ar => ar ! TextMessage(msg.toString))

  def closeConnection(id: UserId) = {
    wsConnections.get(id).foreach { ar =>
      log.info(s"Closing WS connection for $id")
      ar ! Status.Success("Done")
    }
    wsConnections -= id
  }
}

And simple codecs using json circe without any implicits:

import io.circe._
import io.circe.generic.auto._
import io.circe.parser._
import io.circe.syntax._
import WsMsg._

object Codec {

  def decodeWsMsg(json: String): Either[Error, In] = {
    decode[WsMsg.In](json)
  }

  def encodeWsMsg(msg: WsMsg.Out): String = {
    msg.asJson.noSpaces
  }
}

I'm using akka 2.5.11 and akka-http 10.1.0.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.