我使用akka http和streams来完成API请求。当请求无效时，我希望返回一个400，如果请求有效，我希望继续计算，并在之后返回结果。我面临的问题是，我从POST请求接收的有效负载是一个源，我无法将其转换为2个流（一个用于有效输入数据，一个用于无效输入数据），并正确完成请求。

path("alarms")(
  post(entity(asSourceOf[String]) { message =>
    val flow = message.via(Flow[String].map((it) => 
         Try(if valid(it) then it else throw Exception("Wrong input"))
    ))
    complete(repository.create(flow).run) // <-- here I only want to pass all events that are valid. For the other events complete(HttpResponse(NotFound, entity = "Invalid input")) should be used
  })
)
/// The signature of the repository.create looks like that
def create(message: Source[String, NotUsed]): RunnableGraph[Future[Done]]