Akka HTTP 客户端 websocket 意外关闭答案

【问题标题】：Akka HTTP client side websocket closes unexpectedlyAkka HTTP 客户端 websocket 意外关闭
【发布时间】：2019-05-10 13:13:52
【问题描述】：

我有一个 websocket 端点，它每秒向客户端发送一条文本消息。客户端从不向服务器发送任何消息。

使用下面的 JS 代码，它按预期工作，它保持每秒注销消息：

var ws = new WebSocket("ws://url_of_my_endpoint");
ws.onmessage = (message) => console.log(message.data);

我想使用 Akka HTTP 在 Scala 中创建一个类似的消费者。我根据official docs 创建了以下代码。

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
import system.dispatcher

val url = "ws://url_of_my_endpoint"

val outgoing: Source[Message, NotUsed] = Source.empty

val webSocketFlow =
  Http().webSocketClientFlow(WebSocketRequest(url))

val printSink: Sink[Message, Future[Done]] =
  Sink.foreach[Message] {
    case message: TextMessage.Strict =>
      println("message received: " + message.text)
    case _  => println("some other message")
  }

val (upgradeResponse, closed) =
  outgoing
    .viaMat(webSocketFlow)(Keep.right)
    .toMat(printSink)(Keep.both)
    .run()

val connected = upgradeResponse.map { upgrade =>
  if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
    Done
  } else {
    throw new RuntimeException(s"Connection failed: ${upgrade.response.status}")
  }
}

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

问题是连接在几秒钟后关闭。有时 1 秒后，有时 3-4 秒后。 JS客户端工作正常，所以我认为问题不在服务器上。

代码有什么问题？应该怎么改，这样才能告诉我哪里出错了？

【问题讨论】：

标签： scala websocket akka-http

【解决方案1】：

来自documentation：

注意

不活动的 WebSocket 连接将根据idle-timeout settings 被丢弃。如果您需要保持非活动连接处于活动状态，您可以调整空闲超时或定期注入“保持活动”消息。

问题是您没有通过流发送任何消息，因此关闭了非活动连接：

val outgoing: Source[Message, NotUsed] = Source.empty

试试下面的方法，每秒发送一个随机的TextMessage：

import scala.concurrent.duration._

val outgoing: Source[Message, NotUsed] =
  Source
    .fromIterator(() => Iterator.continually(TextMessage(scala.util.Random.nextInt().toString)))
    .throttle(1, 1 second)

或者，调整上述空闲超时设置或配置automatic keep-alive support：

通过将akka.http.client.websocket.periodic-keep-alive-max-idle = 1 second 设置为指定的最大空闲超时，通过配置以透明的方式支持这一点。当在这样的配置期间没有其他消息在传输中时，keep alive 触发。然后，Akka HTTP 将为每个这样的空闲时间间隔自动发送一个Ping frame。

默认情况下，自动保持活动功能处于禁用状态。

【讨论】：

它确实解决了问题。但是，这似乎是一种解决方法；我真的必须发送消息才能保持连接吗？
你在这里得到过答案吗？好像我遇到了同样的问题....

【解决方案2】：

自您查看文档以来，文档可能已更改，因为现在有一个部分可以处理您遇到的问题： https://doc.akka.io/docs/akka-http/current/client-side/websocket-support.html#half-closed-websockets

它解释了：

Akka HTTP WebSocket API 不支持半关闭连接，这意味着如果任一流完成，整个连接将关闭（在交换“关闭握手”或 3 秒超时后）。这可能会导致意外行为，例如，如果我们试图只使用来自服务器的消息

所以这条线

val outgoing: Source[Message, NotUsed] = Source.empty

导致问题。并且可以用以下永远不会完成的行来修复（除非您完成链接到Source.maybe 的Promise）

val outgoing = Source.empty.concatMat(Source.maybe[Message])(Keep.right)

我自己也遇到了这个问题，发现这种行为非常令人困惑。

【讨论】：