如何为Akka Http Websocket连接获取每个用户的终止开关？

import akka.http.scaladsl.model.ws.{Message, TextMessage}
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.Route
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
import akka.stream.{Materializer, OverflowStrategy}

import scala.concurrent.Future

// 定义一个处理WebSocket消息的函数
def handleMessage(message: Message): Option[Message] = message match {
  case TextMessage.Strict(text) => Some(TextMessage("Received: " + text))
  case _ => None
}

// 定义一个处理WebSocket连接的流
def websocketFlow(user: String)(implicit mat: Materializer): Flow[Message, Message, Future[akka.Done]] = {
  val (queue, source) = Source.queue[Message](100, OverflowStrategy.fail)
    .map(handleMessage)  // 处理消息
    .collect { case Some(msg) => msg }  // 过滤掉无效消息
    .toMat(Sink.asPublisher(false))(Keep.both)
    .run()

  // 注册关闭连接时的回调函数
  val terminationSwitch = source.watchCompletion()
  terminationSwitch.foreach(_ => {
    // 连接关闭时的操作
    println(s"User $user disconnected")
  })

  // 将消息流和回调函数生成的终止开关合并为一个Flow对象
  Flow.fromSinkAndSource(Sink.foreach(println), Source.fromPublisher(queue))
}

// 定义一个处理WebSocket连接的路由
def websocketRoute(user: String)(implicit mat: Materializer): Route = {
  path("websocket") {
    handleWebSocketMessages(websocketFlow(user))
  }
}

// 调用websocketRoute方法，传入用户标识和Materializer实例，将其应用于路由系统
val route: Route = websocketRoute("user1")(materializer)