从 Play Websocket 迁移到 Akka HTTP Websocket

Question

我正在从 Play 迁移到 Akka HTTP。我有 jar 依赖代码，我无法更改它接受

Flow[Array[Byte],Array[Byte],Any] 

这是 Play 为 WebSocket 连接提供的。 在 Akka HTTP 中，定义是

Flow[Message,Message,Any] 

我需要两个定义之间的翻译。我是 Akka http 的新手，所以我不确定如何进行。在游戏中，我也在使用 ActorFlow.actorRef

handleWebSocketMessages(wsFlow)

def wsFlow: Flow[Message, Message, Any] = {
 ActorFlow.actorRef(websocket => MyBridgeActor.props(websocket))
}

ActorFlow 代码仅依赖于 akka，因此我刚刚将该文件复制到我自己的代码库中。 https://github.com/playframework/playframework/blob/master/framework/src/play-streams/src/main/scala/play/api/libs/streams/ActorFlow.scala

我想一个解决方案是创建一个 CustomActorFlow，其中包括从 Message 到 Array[Byte] 的转换。 MyBridgeActor 接受 Flow[Array[Byte],Array[Byte],Any] 格式的 websocket。

Answer 1

使用akka stream api，你可以像这样转换流：

import akka.http.scaladsl.model.ws.{BinaryMessage, Message, TextMessage}
import akka.stream.Materializer
import akka.stream.scaladsl.{Flow, Sink}
import akka.util.ByteString

import scala.concurrent.Future

handleWebSocketMessages(msgFlow)

def msgFlow: Flow[Message, Message, Any] = convertFlow(bytesFlow)

def bytesFlow: Flow[Array[Byte], Array[Byte], Any] = {
  // Can just copy ActorFlow over, no need to customize
  ActorFlow.actorRef[Array[Byte], Array[Byte]](...)
}

def covertFlow(msgs: Flow[Array[Byte], Array[Byte], Any])(implicit materializer: Materializer): Flow[Message, Message, Any] =
  Flow[Message]
    .mapAsync(2)(msgToBytes)
    .via(msgs)
    .map(bytesToMsg)


def bytesToMsg(bytes: Array[Byte]): Message = {
  // This depends on your application:
  //   is the outgoing message text or binary?
  val isText = true
  if (isText) {
    TextMessage(new String(bytes, "UTF-8"))
  } else {
    BinaryMessage(ByteString(bytes))
  }
}

def msgToBytes(msg: Message)(implicit materializer: Materializer): Future[Array[Byte]] = {
  msg match {
    case TextMessage.Strict(data) =>
      Future.successful(data.getBytes("UTF-8"))
    case TextMessage.Streamed(stream) =>
      stream.fold("")(_ + _).map(_.getBytes("UTF-8")).runWith(Sink.head)
    case BinaryMessage.Strict(data) =>
      Future.successful(data.toArray[Byte])
    case BinaryMessage.Streamed(stream) =>
      stream.fold(ByteString.empty)(_ ++ _).map(_.toArray[Byte]).runWith(Sink.head)
  }
}