InputStream 在图中关闭运行

Question

我有一个处理大文件的图表。通过从输入流创建的迭代器访问该文件。

for {
  in <- managed(new FileInputStream(inputFile))
} {
  // 3rd party lib that creates iterator from input stream
  val iterator = () => EntityIterator.fromPbf(in) 
  Source
    .fromIterator[OSMEntity](iterator)
    .runForeach(println)
}

（这里我使用的是 ARM，但如果需要，我很乐意手动滚动输入流管理）。

我希望此代码打印迭代器中的每个对象。相反，它会立即完成，因为 in 会在 ARM 块关闭后立即关闭。因为runForeach不阻塞，直接关闭

如果我不使用ARM，我应该什么时候手动关闭输入流？还是我不需要，因为 Akka 会在达到 EOF 时执行此操作？

Answer 1

我没用过scala-arm，不过记得runForeachreturns一个Future[Done]。您可以使用 onComplete.

处理干净

val buf = scala.io.Source.fromFile("readme.md")
Source.fromIterator(() => buf)
  .runForeach(print)
  .onComplete {
    case Success(_) => buf.close()
    case Failure(err) => println(s"error! ${err.getMessage}")
  }

你也应该考虑 Akka 的 FileIO！

FileIO.fromPath(Paths.get("readme.md"))
  .map(_.utf8String)
  .runForeach(print)
  .onComplete {
    case Success(res) => println("done!")
    case Failure(err) => println(err.getMessage)
  }

InputStream 在图中关闭运行

InputStream closing in graph run

akka

akka-stream

InputStream 在图中关闭 运行

InputStream closing in graph run

akka

akka-stream

InputStream 在图中关闭运行