如何让我的传入 websocket 连接始终保持打开状态？

Question

我使用此示例代码客户端连接到我的 websocket 服务，但目前它只是连接然后关闭。

我怎样才能保持这个连接打开并且永远不会关闭它？

一旦建立连接，我希望它保持打开状态，直到我关闭应用程序。

package docs.http.scaladsl

import akka.actor.ActorSystem
import akka.Done
import akka.http.scaladsl.Http
import akka.stream.scaladsl._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws._

import scala.concurrent.Future

object WebSocketClientFlow {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem()
    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)
        case _ =>
        // ignore other message types
      }

    // send this as a message over the WebSocket
    val outgoing = Source.single(TextMessage("hello world!"))

    // flow to use (note: not re-usable!)
    val webSocketFlow = Http().webSocketClientFlow(WebSocketRequest("ws://echo.websocket.org"))

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

Github 参考：https://github.com/akka/akka-http/blob/v10.2.6/docs/src/test/scala/docs/http/scaladsl/WebSocketClientFlow.scala

更新 我的代码与上面相同，只是我更新了我的源代码，如下所示：

val source1 = Source.single(TextMessage("""{"action":"auth","params":"APIKEY_123"}"""))
val source2 = Source.single(TextMessage("""{"action":"subscribe","params":"topic123"}"""))

val sources: Source[Message, NotUsed] =
  Source.combine(source1, source2, Source.maybe)(Concat(_))

所以我可以看到我的 source1 和 source2 被发送到 websocket，但 websocket 并没有开始流式传输其值，它只是挂起。

不知道我做错了什么......

Answer 1

Akka 文档指出了您的情况：

\n

\n

Akka HTTP WebSocket API 不支持半关闭连接，这意味着如果任一流完成，则整个连接将关闭（在交换 \xe2\x80\x9cClosing Handshake\xe2\x80\x9d 后或 3 秒超时后）通过）。

\n

\n

在您的情况下，outgoing（作为 a Source.single）一旦发出 就完成TextMessage。这webSocketFlow完成消息，然后断开连接。

\n

解决办法是推迟outgoing完成，甚至可能永远延迟（或至少直到应用程序被终止）。

\n

在您不想通过 websocket 发送消息的情况下，两个标准源对于延迟完成可能很有用。

\n

\n

• Source.maybe具体化为Promise，您可以使用可选的终止消息来完成。除非承诺完成，否则它不会完成。

  \n
\n

• Source.never永远不会完成。你可以通过不完成来实现这一点Source.maybe，但这比这更少的开销。

  \n
\n

\n

那么它在代码中会是什么样子呢？

\n

val outgoing =\n  Source.single(TextMessage("hello world!"))\n    .concat(Source.never)\n

\n

对于Source.maybe，您需要.concatMat使Promise可以完成；这确实意味着你会得到类似的东西val (completionPromise, upgradeResponse, closed)总体物化值的东西：

\n

val outgoing =\n  Source.single(TextMessage("hello world!"))\n    .concatMat(Source.maybe[TextMessage])(Keep.right)\n\nval ((completionPromise, upgradeResponse), closed) =\n  outgoing\n    .viaMat(websocketFlow)(Keep.both)\n    .toMat(incoming)(Keep.both)\n    .run()\n

\n

在您想要通过套接字发送任意多条消息的情况下，Source.actorRef或者Source.queue很方便：将消息发送到具体化的 actor ref 以通过 websocket 连接发送它们（发送特殊消息以完成源）或将offer消息发送到队列，然后complete它。

\n

val outgoing =\n  Source.actorRef[TextMessage](\n    completionMatcher = {\n      case Done =>\n        CompletionStrategy.draining // send the messages already sent before completing\n    },\n    failureMatcher = PartialFunction.empty,\n    bufferSize = 100,\n    overflowStrategy = OverflowStrategy.dropNew\n  )\n\nval ((sendToSocketRef, upgradeResponse), closed) =\n  outgoing\n    .viaMat(websocketFlow)(Keep.both)\n    .toMat(incoming)(Keep.both)\n    .run()\n\nsendToSocketRef ! TextMessage("hello world!")\n

\n