scala - 如何在 Play 中使用 Akka Streams 将消息从 websocket 客户端发送到 Akka Source

Question

我有一个使用 Play Akka WebSocket 流式传输数字的 Akka Source。websocket 客户端可以发送过滤消息如 0 或 1 来获取偶数或奇数消息。

我不明白如何将过滤器发送到源。

感谢任何反馈

def numbersSource(filter: Int) =
  Source(1 to 1000)
  .throttle(1, 1.second, 1, ThrottleMode.Shaping)
  .filter{ e =>
    filter match {
      case 0 =>
        e % 2 == 0
      case 1 =>
        e % 2 != 0
      case _ =>
        true
    }
  }.map(_.toString)

def testFlow(): Flow[String, String, NotUsed] = {
    Flow[String]
       .filter { msg =>
           //message from ws client
           println(">>"+ msg)
          //validate
          msg != null && !msg.isEmpty
       }
       .via(Flow.fromSinkAndSourceCoupled(Sink.ignore, numbersSource(1)))
  }

def testWSFlow ():WebSocket = {

  WebSocket.acceptOrResult[String, String] {

     case rh if sameOriginCheck(rh) =>
         Future.successful(
      Right(testFlow()))

     case rejected =>
      LOGGER.error(s"Request $rejected failed same origin check")
       Future.successful {
         Left(Forbidden("forbidden"))
      }
    }
 }

标签： scalaplayframeworkakkaakka-streamakka-http