У меня есть следующий фрагмент кода, который не компилируется:
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()
не возвращает Future[WebSocketUpgradeResponse]
в первой позиции в кортеже, но вместо этого возвращает NotUsed
.
Вопрос в том, как получить тип возврата Future[WebSocketUpgradeResponse]
, чтобы определить, что соединение было успешным.