如何在流结束后关闭文件？

Question

我有一个java.io.File序列流。我使用flatMapConcat来创建新的文件Source，如下所示：

def func(files: List[File]) =
  Source(files)
    .map(f => openFile(f))
    .flatMapConcat(f => Source.fromPublisher(SomePublisher(f)))
    .grouped(10)
    .via(SomeFlow)
    .runWith(Sink.ignore)

是否有一种简单的方法在流结束后关闭每个文件？SomePublisher()无法关闭它。

Answer 1

所以我找到了一个很好的方法来解决我的问题，但是如果你有其他的方法，我也想看到它。

def someSource(file: File) = {
  val f = openFile(file)

  Source
    .fromPublisher(SomePublisher(f))
    .transform(() => new PushStage[?, ?] {
      override def onPush(elem: ?, ctx: Context[?]): SyncDirective = ctx.push(elem)

      override def postStop(): Unit = {
        f.close()
        super.postStop()
      }
    }
}

def func(files: List[File]) =
  Source(files)
    .flatMapConcat(someSource)
    .grouped(10)
    .via(SomeFlow)
    .runWith(Sink.ignore)

Answer 2

因此，如果我正确地理解了您的意思，您将执行以下操作:对于创建数据库对象的每个文件，都会打开该文件。因此，由于您在代码中打开db连接，所以您负责关闭它。由于您正在处理有限的文件列表，您可以按顺序存储所有db连接，运行流并在流结束后关闭所有连接。

另一种方法是让您自己的发布者获得文件名，打开db连接，从它获得流，关闭db连接。第二个选项将允许您从无限的文件列表中流。

如果你想要我的代码片段，给我你的函数的完整源代码，我会更新它。