Akka FileIO.fromPath - 如何处理 IOResult 并获取数据？

Question

我看了很多关于这个的例子和帖子。我让它以一种方式工作，但我还没有完全理解这个想法，当我试图将文件读入记录对象流时，我仍然被 Future[IOResult] 绊倒，每行一个, 称它为 Future[List[LineRecordCaseClass]] 是我想要的。

val source = FileIO.fromPath(Paths.get("/tmp/junk_data.csv"))
val flow = makeFlow() // Framing.delimiter->split(",")->map to LineRecordCaseClass
val sink = Sink.collection[LineRecordCaseClass, List[LineRecordCaseClass]]

val graph = source.via(flow).to(sink)

val typeMismatchError: Future[List[LineRecordCaseClass]] = graph.run()

为什么 graph.run() return 变成了 Future[IOResult]？也许我在某处遗漏了 Keep.left 之类的东西？如果有，在什么地方？

我缺少一些概念。

Answer 1

这是你的类型 vals

val source: Source[ByteString, Future[IOResult]] = 
val flow: Flow[ByteString, LineRecordCaseClass, NotUsed] = 
val sink: Sink[LineRecordCaseClass, Future[List[LineRecordCaseClass]]] =

From the akka-stream doc ，在代码片段中

By default, the materialized value of the leftmost stage is preserved

你最左边阶段（源）的物化值是Future[IOResult]。

在 source.via(flow).to(sink) 中，如果您查看 .to 的实现，它会调用 .toMat，默认值为 Keep.left

Keep.both 的类型是

val check: RunnableGraph[(Future[IOResult], Future[List[LineRecordCaseClass]])] = source.via(flow).toMat(sink)(Keep.both)

所以如果你想要Future[List[LineRecordCaseClass]]，你可以做到

source.via(flow).toMat(sink)(Keep.right)

我推荐这个video，它解释了具体化的价值

Akka FileIO.fromPath - 如何处理 IOResult 并获取数据？

Akka FileIO.fromPath - How to deal with IOResult and get the data instead?

scala

akka-stream