异步终止Akka-Http Web套接字连接

Question

lazy val authSuccessMessage = Source.fromFuture(someApiCall)

lazy val messageFlow = requestResponseFlow
    .merge(updateBroadcastEventSource)

lazy val handler = codec
  .atop(authGate(authSuccessMessage))
  .join(messageFlow)

handleWebSocketMessages {
  handler
}

Answer 1

找到了如何使用killswitch来完成此操作。

更新版本

旧版本的问题是，当由堆栈中较高的bidiflow阶段（如我的authgate)触发时，它似乎无法工作。我不确定确切的原因，但是将关闭建模为bidiflow本身，放在堆栈的更高位置，解决了这个问题。

val shutoffPromise = Promise[Option[OutgoingWebsocketEvent]]()

/**
 * Shutoff valve for the connection. It is triggered when `shutoffPromise`
 * completes, and sends a final optional termination message if that
 * promise resolves with one.
 */
val shutoffBidi = {
  val terminationMessageSource = Source
    .maybe[OutgoingWebsocketEvent]
    .mapMaterializedValue(_.completeWith(shutoffPromise.future))

  val terminationMessageBidi = BidiFlow.fromFlows(
    Flow[IncomingWebsocketEventOrAuthorize],
    Flow[OutgoingWebsocketEvent].merge(terminationMessageSource)
  )

  val terminator = BidiFlow
    .fromGraph(KillSwitches.singleBidi[IncomingWebsocketEventOrAuthorize, OutgoingWebsocketEvent])
    .mapMaterializedValue { killSwitch =>
      shutoffPromise.future.foreach { _ => println("Shutting down connection"); killSwitch.shutdown() }
    }

  terminationMessageBidi.atop(terminator)
}

val handler = codec
  .atop(shutoffBidi)
  .atop(authGate(authSuccessMessage))
  .join(messageFlow)

val shutoffPromise = Promise[Option[OutgoingWebsocketEvent]]()

/**
 * Shutoff valve for the flow of outgoing messages. It is triggered when
 * `shutoffPromise` completes, and sends a final optional termination
 * message if that promise resolves with one.
 */
val shutoffFlow = {
  val terminationMessageSource = Source
    .maybe[OutgoingWebsocketEvent]
    .mapMaterializedValue(_.completeWith(shutoffPromise.future))

  Flow
    .fromGraph(KillSwitches.single[OutgoingWebsocketEvent])
    .mapMaterializedValue { killSwitch =>
      shutoffPromise.future.foreach(_ => killSwitch.shutdown())
    }
    .merge(terminationMessageSource)
}

val handler = codec
  .atop(authGate(authSuccessMessage))
  .join(messageFlow via shutoffFlow)