Akka流消耗Web套接字 [英] akka stream consume web socket
问题描述
从akka流开始，我想构建一个简单的示例。
在Chrome中使用Web套接字插件，我可以简单地连接到像这样的流 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扩展App {
隐式val system = ActorSystem（）
隐式val实现器= ActorMaterializer（）
导入system.dispatcher
//打印每个传入的严格文本消息
val printSink：Sink [Message，Future [Done ]] =
Sink.foreach {
案例消息：TextMessage.Strict =>
println（message.text）
}
val commandMessages = Seq（TextMessage（ {\ op\：\ ping\}） ，TextMessage（ {\ op\：\ unconfirmed_sub\}）））
val helloSource：Source [消息，未使用] = Source（commandMessages.to [scala.collection.immutable。 Seq]）
// Future [Done]是Sink.foreach
//的物化值，并且在流完成
val流时完成：Flow [Message ，Message，Future [Done]] =
Flow.fromSinkAndSourceMat（printSink，helloSource）（Keep.left）
// upgradeResponse是Future [WebSocketUpgradeResponse]，即
//连接成功或失败时完成或失败
// //关闭是一个Future [Done]，表示从
val（upgradeResponse，关闭）以上的流完成=
Http（）。singleWebSocketRequest（ WebSocketRequest（ wss：//ws.blockchain.info/inv），流）
val connected = upgradeResponse.map {upgra de =>
//就像常规的http请求一样，我们可以访问通过upgrade.response.status
//状态代码101（交换协议）表示的响应状态，表明服务器支持WebSockets
if（ upgrade.response.status == StatusCodes.SwitchingProtocols）{
完成
}否则{
抛出新的RuntimeException（s连接失败：$ {upgrade.response.status}）
}
}
//在实际的应用程序中，您不会在这里
//产生副作用，并且更仔细地处理错误
connected.onComplete（println）// TODO为什么我无法获得与chrome中相同的输出？
close.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}
查看代码：
对象WebSocketClientFlow扩展了App {
隐式val系统= ActorSystem（）
隐式val物料化器= ActorMaterializer（）
导入system.dispatcher
// Future [Done]是Sink.foreach的物化值，
//当流完成
val传入时发出：Sink [Message，Future [Done]] =
Sink.foreach [Message] {
案例消息：TextMessage.Strict =& gt;
println（message.text）
}
//通过WebSocket将其作为消息发送
val commandMessages = Seq（TextMessage（ {\ op \：\ ping\}），TextMessage（ {\ op\：\ unconfirmed_sub\}）））
val out：Source [消息，未使用] = Source（commandMessages.to [scala.collection.immutable.Seq]）
// valout = Source.single（TextMessage（ hello world！））
//流使用（注意：不可重用！）
val webSocketFlow = Http（）。webSocketClientFlow（WebSocketRequest（ wss：//ws.blockchain.info/inv））
/ /物化值是带有
的元组// upgradeResponse是Future [WebSocketUpgradeResponse]，当连接成功或失败时，
//完成或失败
// //关闭是Future [Done ]，并从传入接收器
val的流完成（upgradeResponse，关闭）=
传出
.viaMat（webSocketFlow）（Keep.right）//保留物化的Future [WebSocketUpgradeResponse]
.toMat（incoming）（Keep.both）//还保留Future [Done]
.run（）
//就像常规的http请求一样，我们可以访问响应状态为可通过upgrade.response.status
//状态代码101（交换协议）表示服务器支持WebSockets
val connected = upgradeResponse.flatMap {upgrade =>
if（upgrade.response.status == StatusCodes.SwitchingProtocols）{
Future.successful（Done）
} else {
抛出新的RuntimeException（s连接失败：$ { upgrade.response.status}）
}
}
//在实际的应用程序中，您不会对此产生副作用
connected.onComplete（println）
close.foreach（_ => {
println（ closed）
system.terminate
}）
}
我如何获得与chrome中相同的结果
- 显示已订阅流的打印内容
- 最好定期发送 https://blockchain.info/api/api 通过
{ op： ping}消息
注意，我在版本 2.4.17 中使用akka和版本在 10.0中的akka-http .5
我注意到的几件事e是：
1），您需要使用所有类型的传入消息，而不仅仅是 TextMessage.Strict 种。区块链流绝对是流式消息，因为它包含大量文本，并将通过网络以块的形式传递。更完整的接收器接收器可能是：
val接收器：接收器[消息，将来[完成]] =
流[Message] .mapAsync（4）{
案例消息：TextMessage.Strict =>
println（message.text）
Future.successful（Done）
案例消息：TextMessage.Streamed =>
message.textStream.runForeach（println）
案例消息：BinaryMessage =>
message.dataStream.runWith（Sink.ignore）
} .toMat（Sink.last）（Keep.right）
2），您的2个元素的来源可能太早完成，即在websocket响应返回之前。您可以通过执行
val传出来连接 Source。也许 [Strict，Promise [Option [Nothing]]] =
Source（commandMessages.to [scala.collection.immutable.Seq]）。concatMat（Source.maybe）（Keep.right）
然后
val （completionPromise，upgradeResponse），已关闭）=
传出
.viaMat（webSocketFlow）（Keep.both）
.toMat（incoming）（Keep.both）
.run（）
通过使物化承诺不完整，可以保持源打开并避免流程关闭。 / p>
Getting started with akka-streams I want to build a simple example.
In chrome using a web socket plugin I simply can connect to a stream like this one https://blockchain.info/api/api_websocket via wss://ws.blockchain.info/inv and sending 2 commands
{"op":"ping"}{"op":"unconfirmed_sub"}will stream the results in chromes web socket plugin window.
I tried to implement the same functionality in akka streams but am facing some problems:
- 2 commands are executed, but I actually do not get the streaming output
- the same command is executed twice (the ping command)
When following the tutorial of http://doc.akka.io/docs/akka/2.4.7/scala/http/client-side/websocket-support.html or http://doc.akka.io/docs/akka-http/10.0.0/scala/http/client-side/websocket-support.html#half-closed-client-websockets Here is my adaption below:
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"))
}
when using the flow version from http://doc.akka.io/docs/akka-http/10.0.0/scala/http/client-side/websocket-support.html#websocketclientflow modified as outlined below, again, the result is twice the same output:
{"op":"pong"}
{"op":"pong"}
See the code:
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
})
}
How can I achieve the same result as in chrome
- display print of subscribed stream
- at best periodically send update (ping statements) as outlined in https://blockchain.info/api/api via
{"op":"ping"}messages
Note, I am using akka in version 2.4.17 and akka-http in version 10.0.5
A couple of things I notice are:
1) you need to consume all types of incoming messages, not only the TextMessage.Strict kind. The blockchain stream is definitely a Streamed message, as it contains loads of text and it will be delivered in chunks over the network. A more complete incoming Sink could be:
val incoming: Sink[Message, Future[Done]] =
Flow[Message].mapAsync(4) {
case message: TextMessage.Strict =>
println(message.text)
Future.successful(Done)
case message: TextMessage.Streamed =>
message.textStream.runForeach(println)
case message: BinaryMessage =>
message.dataStream.runWith(Sink.ignore)
}.toMat(Sink.last)(Keep.right)
2) your source of 2 elements might complete too early, i.e. before the websocket responses come back. You can concatenate a Source.maybe by doing
val outgoing: Source[Strict, Promise[Option[Nothing]]] =
Source(commandMessages.to[scala.collection.immutable.Seq]).concatMat(Source.maybe)(Keep.right)
and then
val ((completionPromise, upgradeResponse), closed) =
outgoing
.viaMat(webSocketFlow)(Keep.both)
.toMat(incoming)(Keep.both)
.run()
by keeping the materialized promise non-complete, you keep the source open and avoid the flow shutdown.
这篇关于Akka流消耗Web套接字的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
