从服务器关闭akka-http websocket连接

Question

关闭连接可以通过取消来自服务器逻辑的传入连接流（例如，将其下游连接到sink.canceled并将其上游连接到source.empty）。还可以通过取消IncomingConnection源连接来关闭服务器的套接字。

但考虑到接收器和源会在协商新连接时设置一次，我不清楚如何做到这一点：

(get & path("ws")) {
  optionalHeaderValueByType[UpgradeToWebsocket]() {
    case Some(upgrade) ⇒
      val connectionId = UUID()
      complete(upgrade.handleMessagesWithSinkSource(sink, source))
    case None ⇒
      reject(ExpectedWebsocketRequestRejection)
  }
}

Answer 1

提示：此答案基于Akka-Stream-Experiment版本2.0-M2。其他版本的API可能略有不同。

关闭连接的一种简单方法是使用pushstage:

import akka.stream.stage._

val closeClient = new PushStage[String, String] {
  override def onPush(elem: String, ctx: Context[String]) = elem match {
    case "goodbye" ⇒
      // println("Connection closed")
      ctx.finish()
    case msg ⇒
      ctx.push(msg)
  }
}

在客户端或服务器端接收的每个元素（通常，通过flow的每个元素）都通过这样的stage组件。在Akka中，完整的抽象称为graphstage，更多信息可以在官方文档中找到。

val connection = Tcp().outgoingConnection(address, port)

val flow = Flow[ByteString]
  .via(Framing.delimiter(
      ByteString("\n"),
      maximumFrameLength = 256,
      allowTruncation = true))
  .map(_.utf8String)
  .transform(() ⇒ closeClient)
  .map(_ ⇒ StdIn.readLine("> "))
  .map(_ + "\n")
  .map(ByteString(_))

connection.join(flow).run()