如何使用Akka Streams和Akka HTTP为websockets订阅actor的消息？

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.ws.{Message, TextMessage}
import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Sink, Source}

object WebSocketServer extends App {
  implicit val system = ActorSystem("websocket-server")
  implicit val materializer = ActorMaterializer()
  implicit val executionContext = system.dispatcher

  // 定义一个Actor，用于处理WebSocket消息
  class WebSocketActor extends Actor {
    def receive: Receive = {
      case msg: String =>
        // 处理接收到的消息
        println(s"Received message: $msg")
    }
  }

  // 创建一个WebSocket路由器
  val webSocketRoute = path("websocket") {
    handleWebSocketMessages(webSocketFlow)
  }

  // 创建一个WebSocket流
  val webSocketFlow: Flow[Message, Message, Any] = {
    val actorRef: ActorRef = system.actorOf(Props[WebSocketActor])
    val incomingMessages: Sink[Message, Any] =
      Flow[Message].map {
        case TextMessage.Strict(msg) => msg
      }.to(Sink.actorRef(actorRef, "completed"))

    val outgoingMessages: Source[Message, Any] =
      Source.actorRef[String](bufferSize = 10, OverflowStrategy.fail)
        .mapMaterializedValue { outActor =>
          actorRef ! outActor
          outActor
        }.map(msg => TextMessage.Strict(msg))

    Flow.fromSinkAndSource(incomingMessages, outgoingMessages)
  }

  // 启动HTTP服务器
  val bindingFuture = Http().bindAndHandle(webSocketRoute, "localhost", 8080)
  println("Server started at http://localhost:8080/websocket")

  // 等待服务器终止
  sys.addShutdownHook {
    bindingFuture.flatMap(_.unbind()).onComplete(_ => system.terminate())
  }
}