scala - 使用 Play 2.6 和 akka 流的 Websocket 代理

Question

我正在尝试使用 Play 和 akka 流为 Websocket 连接创建一个简单的代理。
流量是这样的:

(Client) request  ->         -> request (Server)
                      Proxy 
(Client) response <-         <- response (Server)

在遵循一些示例后，我想出了以下代码:

def socket = WebSocket.accept[String, String] { request =>

val uuid = UUID.randomUUID().toString

// wsOut - actor that deals with incoming websocket frame from the Client
// wsIn - publisher of the frame for the Server
val (wsOut: ActorRef, wsIn: Publisher[String]) = {
  val source: Source[String, ActorRef] = Source.actorRef[String](10, OverflowStrategy.dropTail)
  val sink: Sink[String, Publisher[String]] = Sink.asPublisher(fanout = false)
  source.toMat(sink)(Keep.both).run()
}

// sink that deals with the incoming messages from the Server
val serverIncoming: Sink[Message, Future[Done]] =
  Sink.foreach[Message] {
    case message: TextMessage.Strict =>
      println("The server has sent: " + message.text)
  }

// source for sending a message over the WebSocket
val serverOutgoing = Source.fromPublisher(wsIn).map(TextMessage(_))

// flow to use (note: not re-usable!)
val webSocketFlow = Http().webSocketClientFlow(WebSocketRequest("ws://0.0.0.0:6000"))

// the materialized value is a tuple with
// upgradeResponse is a Future[WebSocketUpgradeResponse] that
// completes or fails when the connection succeeds or fails
// and closed is a Future[Done] with the stream completion from the incoming sink
val (upgradeResponse, closed) =
serverOutgoing
  .viaMat(webSocketFlow)(Keep.right) // keep the materialized Future[WebSocketUpgradeResponse]
  .toMat(serverIncoming)(Keep.both) // also keep the Future[Done]
  .run()

// just like a regular http request we can access response status which is available via upgrade.response.status
// status code 101 (Switching Protocols) indicates that server support WebSockets
val connected = upgradeResponse.flatMap { upgrade =>
  if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
    Future.successful(Done)
  } else {
    throw new RuntimeException(s"Connection failed: ${upgrade.response.status}")
  }
}

// in a real application you would not side effect here
connected.onComplete(println)
closed.foreach(_ => println("closed"))

val actor = system.actorOf(WebSocketProxyActor.props(wsOut, uuid))
val finalFlow = {
  val sink = Sink.actorRef(actor, akka.actor.Status.Success(()))
  val source = Source.maybe[String] // what the client receives. How to connect with the serverIncoming sink ???
  Flow.fromSinkAndSource(sink, source)
}

finalFlow

使用此代码，流量从客户端到代理再到服务器，再返回到代理，仅此而已。它不会进一步到达客户端。我怎样才能解决这个问题 ？
我想我需要以某种方式连接 serverIncoming沉到 source在 finalFlow ，但我不知道该怎么做......

还是我对这种方法完全错误？使用 Bidiflow 更好吗？或 Graph ?我是 akka 流的新手，仍在尝试解决问题。

Answer 1

作为 Federico 非常好的解决方案的扩展 - 此代码可用于代理转发网关服务，您可以在其中连接到代理，该代理将 Websockets“管道”到微服务。下面的代码使用 Akka Http 10.2.0，并且代码中有规定在发起者 Websocket 客户端断开连接时处理上游流失败 - 即，通过添加到 Websocket 客户端流的 case throwable 来恢复。

import akka.actor.typed.ActorSystem
import akka.actor.typed.scaladsl.Behaviors
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.ws._
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.server.Directives.{complete, extractWebSocketUpgrade}
import akka.stream.scaladsl._

import scala.io.StdIn
import scala.util.{Failure, Success}

object Main {

  def main(args: Array[String]) {

    implicit val system = ActorSystem(Behaviors.empty, "webtest")
    implicit val executionContext = system.executionContext

    def webSocketFlow =
      Http().webSocketClientFlow(WebSocketRequest("ws://localhost:8000/ws"))
        .recover {
          case throwable: Throwable =>
            try {
              throw new RuntimeException(s"Websocket Upstream Flow failed... Message: ${throwable.getMessage}")
            } catch {
              case t: Throwable => system.log.info(t.getMessage)           //catching all Throwable exceptions
            }
            TextMessage("Websocket Upstream Flow failed...")
        }

    def routeFlow: Flow[HttpRequest, HttpResponse, Any] = extractWebSocketUpgrade { upgrade =>
      val handleWebSocketProxy = upgrade.handleMessages(webSocketFlow)
      complete(handleWebSocketProxy)
    }

    Http().newServerAt("0.0.0.0", 8080).bindFlow(routeFlow)
      .onComplete {
        case Success(_) =>
          system.log.info("Server online at http://0.0.0.0:8080")
        case Failure(ex) =>
          system.log.error("Failed to bind HTTP endpoint, terminating system", ex)
          system.terminate()
      }

    system.log.info("Press RETURN to stop...")
    StdIn.readLine()
    system.terminate()
  }

}

这里用户/发起者作为代理连接到 0.0.0.0:8080 并“通过管道”(转发)到 localhost:8000。