scala - Akka 流 : validation of elements being streamed

Question

我是 Akka Streams 的新手，我想知道如何实现某种中流验证。示例:

FileIO
  .fromPath(file)
  .via(Framing.delimiter(...)
  .map(_.utf8String)
  .map(_.split("\t", -1))
  .validate(arr => arr.length == 10) // or similar
  ...

我假设这种情况如此普遍，以至于必须有一个预定义的功能来动态验证流。但是，我找不到任何相关信息。我是不是走错了路，在 Akka Streams 中不应该以这种方式进行验证？

在我的特定场景中，我正在逐行处理一个文件。如果只有一行无效，则继续没有意义，应该中止处理。

Answer 1

我可能会创建一个类型来表示约束，然后您可以在创建该类型的实例时进行断言，并了解下游应用了哪些约束。

例子:

object LineItem {
  // Makes it possible to provide the validation before allocating the item
  def apply(string: String): LineItem = {
    require(string.length == 10)
    new LineItem(string) // Call the companion-accessible constructor
  }
}
// private[LineItem] makes sure that `new` only works from companion object
final case class LineItem private[LineItem](string: String)