Akka Stream Kansume Web 套接字

Question

开始使用 akka-streams 我想构建一个简单的示例。在 chrome 中，使用网络套接字插件，我可以通过 wss://ws.blockchain.info/inv 连接到像这样的流 https://blockchain.info/api/api_websocket 并发送 2 个命令

{"op":"ping"}
{"op":"unconfirmed_sub"} 将在 chromes 网络套接字插件 window.

我试图在 akka 流中实现相同的功能，但遇到了一些问题：

执行了2条命令，但我实际上没有得到流输出
同一个命令执行两次（ping命令）

当遵循 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 的教程时以下是我的改编：

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
  })
}

如何获得与 chrome

相同的结果

显示订阅流的打印
最好定期发送更新（ping 语句），如 https://blockchain.info/api/api 中所述，通过 {"op":"ping"} 消息

注意，我在 2.4.17 版本中使用 akka，在 10.0.5

版本中使用 akka-http

Answer 1

我注意到的几件事是：

1) 您需要使用所有类型的传入消息，而不仅仅是 TextMessage.Strict 类型。区块链流绝对是流式消息，因为它包含大量文本，并且将通过网络以块的形式传递。一个更完整的传入接收器可能是：

  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) 您的 2 个元素的来源可能完成得太早，即在 websocket 响应返回之前。您可以通过

连接 Source.maybe

val outgoing: Source[Strict, Promise[Option[Nothing]]] =
    Source(commandMessages.to[scala.collection.immutable.Seq]).concatMat(Source.maybe)(Keep.right)

然后

  val ((completionPromise, upgradeResponse), closed) =
    outgoing
      .viaMat(webSocketFlow)(Keep.both)
      .toMat(incoming)(Keep.both)
      .run()

通过保持物化承诺不完整，您可以保持源开放并避免流程关闭。

Akka Stream Kansume Web 套接字

akka stream consume web socket

scala

akka

akka-stream

reactive-streams