当客户端关闭 Web 套接字连接时停止 Akka 流源

Question

我有一个akka http web socket Route，其代码类似于：

private val wsReader: Route = 路径（“v1”/“数据”/“ws”）{ log.info("正在打开 websocket 连接...")

  val testSource = Source
    .repeat("Hello")
    .throttle(1, 1.seconds)
    .map(x => {
      println(x)
      x
    })
    .map(TextMessage.Strict)
    .limit(1000)

  extractUpgradeToWebSocket { upgrade ⇒
    complete(upgrade.handleMessagesWithSinkSource(Sink.ignore, testSource))
  }
}

一切正常（我每秒从客户端收到 1 条测试消息）。唯一的问题是，如果客户端关闭 Web 套接字连接，我不明白如何停止/关闭 Source (testSource)。

如果网络套接字关闭，您也可以看到源继续生成元素（请参阅println）。

如何检测客户端断开连接？

Answer 1

handleMessagesWithSinkSource 实现为：

/**
 * The high-level interface to create a WebSocket server based on "messages".
 *
 * Returns a response to return in a request handler that will signal the
 * low-level HTTP implementation to upgrade the connection to WebSocket and
 * use the supplied inSink to consume messages received from the client and
 * the supplied outSource to produce message to sent to the client.
 *
 * Optionally, a subprotocol out of the ones requested by the client can be chosen.
 */
def handleMessagesWithSinkSource(
  inSink:      Graph[SinkShape[Message], Any],
  outSource:   Graph[SourceShape[Message], Any],
  subprotocol: Option[String]                   = None): HttpResponse =

  handleMessages(Flow.fromSinkAndSource(inSink, outSource), subprotocol)

这意味着 sink 和 source 是独立的，事实上即使客户端关闭了连接的传入端，source 也应该继续生成元素。不过，当客户端完全重置连接时，它应该会停止。

要在传入连接关闭后立即停止生成传出数据，您可以使用Flow.fromSinkAndSourceCoupled，因此：

val socket = upgrade.handleMessages(
  Flow.fromSinkAndSourceCoupled(inSink, outSource)
  subprotocol = None
)

Answer 2

一种方法是使用KillSwitches 处理 testSource 关闭。

private val wsReader: Route =
path("v1" / "data" / "ws") {
  logger.info("Opening websocket connecting ...")

  val sharedKillSwitch = KillSwitches.shared("my-kill-switch")

  val testSource =
    Source
     .repeat("Hello")
     .throttle(1, 1.seconds)
     .map(x => {
       println(x)
       x
     })
    .map(TextMessage.Strict)
    .limit(1000)
    .via(sharedKillSwitch.flow)

  extractUpgradeToWebSocket { upgrade ⇒
    val inSink = Sink.onComplete(_ => sharedKillSwitch.shutdown())
    val outSource = testSource
    val socket = upgrade.handleMessagesWithSinkSource(inSink, outSource)

    complete(socket)
  }
}