Приступая к работе с akka-streams, я хочу построить простой пример. В Chrome с помощью подключаемого модуля веб-сокета я просто могу подключиться к потоку, подобному этому https://blockchain.info/api/api_websocket через wss://ws.blockchain.info/inv
и отправив 2 команды
{"op":"ping"}
{"op":"unconfirmed_sub"}
будет передавать результаты в окно плагина веб-сокета chromes.
Я попытался реализовать ту же функциональность в потоках akka, но столкнулся с некоторыми проблемами:
- Выполняются 2 команды, но на самом деле я не получаю потокового вывода
- одна и та же команда выполняется дважды (команда ping)
Следуя руководству http://doc.akka.io/docs/akka/2.4.7/scala/http/client-side/websocket-support.html или http://doc.akka.io/docs/akka-http/10.0.0/scala/http/client-side/websocket-support.html#half-closed-client-websockets Вот моя адаптация ниже:
object SingleWebSocketRequest extends App {
implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
import system.dispatcher
// print each incoming strict text message
val printSink: Sink[Message, Future[Done]] =
Sink.foreach {
case message: TextMessage.Strict =>
println(message.text)
}
val commandMessages = Seq(TextMessage("{\"op\":\"ping\"}"), TextMessage("{\"op\":\"unconfirmed_sub\"}"))
val helloSource: Source[Message, NotUsed] = Source(commandMessages.to[scala.collection.immutable.Seq])
// the Future[Done] is the materialized value of Sink.foreach
// and it is completed when the stream completes
val flow: Flow[Message, Message, Future[Done]] =
Flow.fromSinkAndSourceMat(printSink, helloSource)(Keep.left)
// upgradeResponse is a Future[WebSocketUpgradeResponse] that
// completes or fails when the connection succeeds or fails
// and closed is a Future[Done] representing the stream completion from above
val (upgradeResponse, closed) =
Http().singleWebSocketRequest(WebSocketRequest("wss://ws.blockchain.info/inv"), flow)
val connected = upgradeResponse.map { upgrade =>
// 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
if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
Done
} else {
throw new RuntimeException(s"Connection failed: ${upgrade.response.status}")
}
}
// in a real application you would not side effect here
// and handle errors more carefully
connected.onComplete(println) // TODO why do I not get the same output as in chrome?
closed.foreach(_ => println("closed"))
}
при использовании версии потока из http://doc.akka.io/docs/akka-http/10.0.0/scala/http/client-side/websocket-support.html#websocketclientflow измененный, как описано ниже, опять же, результат - это в два раза тот же результат:
{"op":"pong"}
{"op":"pong"}
Смотрите код:
object WebSocketClientFlow 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)
}
// send this as a message over the WebSocket
val commandMessages = Seq(TextMessage("{\"op\":\"ping\"}"), TextMessage("{\"op\":\"unconfirmed_sub\"}"))
val outgoing: Source[Message, NotUsed] = Source(commandMessages.to[scala.collection.immutable.Seq])
// val outgoing = Source.single(TextMessage("hello world!"))
// flow to use (note: not re-usable!)
val webSocketFlow = Http().webSocketClientFlow(WebSocketRequest("wss://ws.blockchain.info/inv"))
// 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) =
outgoing
.viaMat(webSocketFlow)(Keep.right) // keep the materialized Future[WebSocketUpgradeResponse]
.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}")
}
}
// in a real application you would not side effect here
connected.onComplete(println)
closed.foreach(_ => {
println("closed")
system.terminate
})
}
Как добиться того же результата, что и в хроме
- отображать распечатку подписанного потока
- в лучшем случае периодически отправляйте обновления (запросы ping), как указано в https://blockchain.info/api/api через
{"op":"ping"}
messages
Обратите внимание: я использую akka в версии 2.4.17
и akka-http в версии 10.0.5
.