协议切换成功与否

Question

我有以下无法编译的代码片段：

import akka.actor.ActorSystem
import akka.Done
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws._

import scala.concurrent._
import scala.concurrent.duration._


object Main extends App {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  import system.dispatcher

  // Future[Done] is the materialized value of Sink.foreach,
  // emitted when the stream completes
  val incoming: Sink[Message, Future[Done]] =
  Sink.foreach[Message] {
    case message: TextMessage.Strict =>
      println(message.text)
  }


  // flow to use (note: not re-usable!)
  val sourceSocketFlow = RestartSource.withBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2,
    maxRestarts = 3
  ) { () =>
    Source.tick(2.seconds, 2.seconds, TextMessage("Hello world!!!"))
        .viaMat(Http().webSocketClientFlow(WebSocketRequest("ws://127.0.0.1:8080/")))(Keep.right)
  }
  // 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) =
    sourceSocketFlow
    .toMat(incoming)(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}")
    }
  }

  connected.onComplete(println)
  closed.foreach(_ => println("closed"))

}

这里的问题是：

  // 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) =
    sourceSocketFlow
    .toMat(incoming)(Keep.both) // also keep the Future[Done]
    .run()

不会 return Future[WebSocketUpgradeResponse] 在元组的第一个位置，而是 returns NotUsed。

问题是，如何让return类型Future[WebSocketUpgradeResponse]识别，连接成功。

Answer 1

RestartSource#withBackoff 接受类型 () => Source[T, _] 的 sourceFactory 和 returns 类型 Source[T, NotUsed] 的新来源。所以不可能从包装的源中提取物化值。这可能是因为每次重新启动 RestartSource 时物化值都会不同。

The question is, how to get the return type Future[WebSocketUpgradeResponse] to identify, that the connection was successful.

如果您想检查连接是否已建立，以及 WebSockets 握手是否成功，您可以使用 Source#preMaterialize 预先实现源。您的代码稍微修改后的版本如下所示：

val sourceSocketFlow: Source[Message, NotUsed] = RestartSource.withBackoff(
  minBackoff = 3.seconds,
  maxBackoff = 30.seconds,
  randomFactor = 0.2,
  maxRestarts = 3
) { () =>
  val (response, source) = Source
    .tick(2.seconds, 2.seconds, TextMessage("Hello world!!!"))
    .viaMat(Http().webSocketClientFlow(WebSocketRequest("ws://mockbin.org/bin/82b160d4-6c05-4943-908a-a15122603e20")))(Keep.right).preMaterialize()

  response.onComplete {
    case Failure(e) ⇒
      println(s"Connection failed")

    case Success(value) ⇒
      if (value.response.status == StatusCodes.SwitchingProtocols) {
        println("Server supports websockets")
      } else {
        println("Server does not support websockets")
      }
  }

  source
}

如果连接失败，或 websocket handshake fails，您无需执行任何操作。这两种情况都将由 RestartSource 处理。

协议切换成功与否

Protocol switching successfully or not

scala

akka

akka-stream

akka-http