我想为目录和子目录中的每个文件应用一个函数,如下所示:
def applyRecursively(dir: String, fn: (File) => Any) {
def listAndProcess(dir: File) {
dir.listFiles match {
case null => out.println("exception: dir cannot be listed: " + dir.getPath); List[File]()
case files => files.toList.sortBy(_.getName).foreach(file => {
fn(file)
if (!java.nio.file.Files.isSymbolicLink(file.toPath) && file.isDirectory) listAndProcess(file)
})
}
}
listAndProcess(new File(dir))
}
def exampleFn(file: File) { println(s"processing $file") }
applyRecursively(dir, exampleFn)
这是有效的。这里的问题是如何使用scalaIteratees重构这段代码。像这样的东西:
val en = Enumerator.generateM(...) // ???
val it: Iteratee[File, Unit] = Iteratee.foreach(exampleFn)
val res = en.run(it)
res.onSuccess { case x => println("DONE") }
您可以为此使用Enumerator.unfold
。签名为:
def unfold[S, E](s: S)(f: (S) => Option[(S, E)])(implicit ec: ExecutionContext): Enumerator[E]
其思想是从S
类型的值开始,然后对其应用一个返回Option[(S, E)]
的函数。值None
表示Enumerator
已经达到EOF。一个Some
包含另一个要展开的S
,以及Enumerator[E]
将生成的下一个值。在您的示例中,您可以从Array[File]
(初始目录)开始,从Array
中获取第一个值,然后检查它是文件还是目录。如果它只是一个文件,则返回Array
的尾部,并将File
元组在一起。如果File
是一个目录,则获得文件列表并将其添加到Array
的开头。然后unfold
中的下一步将继续处理包含的文件。
你最终会得到这样的东西:
def list(dir: File)(implicit ec: ExecutionContext): Enumerator[File] = {
Enumerator.unfold(Array(dir)) { listing =>
listing.headOption.map { file =>
if(!java.nio.file.Files.isSymbolicLink(file.toPath) && file.isDirectory)
(file.listFiles.sortBy(f => (f.isDirectory, f.getName)) ++ listing.tail) -> file
else
listing.tail -> file
}
}
}
我通过isDirectory
添加了一个额外的排序,以优先考虑非目录。这意味着,如果将目录内容添加到Array
以展开,则在添加更多内容之前,将首先消耗文件。这将防止内存占用由于递归性质而快速扩展。
如果希望从最终的Enumerator
中删除目录,可以使用Enumeratee.filter
来执行此操作。你最终会得到这样的东西:
list(dir) &> Enumeratee.filter(!_.isDirectory) |>> Iteratee.foreach(fn)
它不会捕获您的所有需求,但这可以让您启动
object ExampleEnumerator {
import scala.concurrent.ExecutionContext.Implicits.global
def exampleFn(file: File) { println(s"processing $file") }
def listFiles(dir: File): Enumerator[File] = {
val files = Option(dir.listFiles).toList.flatten.sortBy(_.getName)
Enumerator(dir) andThen Enumerator(files :_*).flatMap(listFiles)
}
def main(args: Array[String]) {
import scala.concurrent.duration._
val dir = "."
val en: Enumerator[File] = listFiles(new File(dir))
val it: Iteratee[File, Unit] = Iteratee.foreach(exampleFn)
val res = en.run(it)
res.onSuccess { case x => println("DONE") }
Await.result(res, 10.seconds)
}
}
这只是用一些日志记录来补充m-w
的伟大答案,以帮助理解它
$ cd /david/test
$ find .
.
./file1
./file2
./file3d
./file3d/file1
./file3d/file2
./file4
java:
import play.api.libs.iteratee._
import java.io.File
import scala.concurrent.Await
import scala.concurrent.duration.Duration
object ExampleEnumerator3 {
import scala.concurrent.ExecutionContext.Implicits.global
def exampleFn(file: File) { println(s"processing $file") }
def list(dir: File): Enumerator[File] = {
println(s"list $dir")
val initialInput: List[File] = List(dir)
Enumerator.unfold(initialInput) { (input: List[File]) =>
val next: Option[(List[File], File)] = input.headOption.map { file =>
if(file.isDirectory) {
(file.listFiles.toList.sortBy(_.getName) ++ input.tail) -> file
} else {
input.tail -> file
}
}
next match {
case Some(dn) => print(s"value to unfold: $inputn next value to unfold: ${dn._1}n next input: ${dn._2}n")
case None => print(s"value to unfold: $inputn finished unfoldn")
}
next
}
}
def main(args: Array[String]) {
val dir = new File("/david/test")
val res = list(dir).run(Iteratee.foreach(exampleFn))
Await.result(res, Duration.Inf)
}
}
日志:
list /david/test
value to unfold: List(/david/test)
next value to unfold: List(/david/test/file1, /david/test/file2, /david/test/file3d, /david/test/file4)
next input: /david/test
processing /david/test
value to unfold: List(/david/test/file1, /david/test/file2, /david/test/file3d, /david/test/file4)
next value to unfold: List(/david/test/file2, /david/test/file3d, /david/test/file4)
next input: /david/test/file1
processing /david/test/file1
value to unfold: List(/david/test/file2, /david/test/file3d, /david/test/file4)
next value to unfold: List(/david/test/file3d, /david/test/file4)
next input: /david/test/file2
processing /david/test/file2
value to unfold: List(/david/test/file3d, /david/test/file4)
next value to unfold: List(/david/test/file3d/file1, /david/test/file3d/file2, /david/test/file4)
next input: /david/test/file3d
processing /david/test/file3d
value to unfold: List(/david/test/file3d/file1, /david/test/file3d/file2, /david/test/file4)
next value to unfold: List(/david/test/file3d/file2, /david/test/file4)
next input: /david/test/file3d/file1
processing /david/test/file3d/file1
value to unfold: List(/david/test/file3d/file2, /david/test/file4)
next value to unfold: List(/david/test/file4)
next input: /david/test/file3d/file2
processing /david/test/file3d/file2
value to unfold: List(/david/test/file4)
next value to unfold: List()
next input: /david/test/file4
processing /david/test/file4
value to unfold: List()
finished unfold
这只是用一些日志记录来补充@JonasAnso的伟大答案,以帮助理解它。
$ cd /david/test
$ find .
.
./file1
./file2
./file3d
./file3d/file1
./file3d/file2
./file4
java:
import play.api.libs.iteratee._
import java.io.File
import scala.concurrent.Await
import scala.concurrent.duration.Duration
object ExampleEnumerator2b {
import scala.concurrent.ExecutionContext.Implicits.global
def exampleFn(file: File) { println(s"processing $file") }
def listFiles(dir: File): Enumerator[File] = {
println(s"listFiles. START: $dir")
if (dir.isDirectory) {
val files = dir.listFiles.toList.sortBy(_.getName)
Enumerator(dir) andThen Enumerator(files :_*).flatMap(listFiles)
} else {
Enumerator(dir)
}
}
def main(args: Array[String]) {
val dir = new File("/david/test2")
val res = listFiles(dir).run(Iteratee.foreach(exampleFn))
Await.result(res, Duration.Inf)
}
}
日志:
listFiles. START: /david/test
processing /david/test
listFiles. START: /david/test/file1
processing /david/test/file1
listFiles. START: /david/test/file2
processing /david/test/file2
listFiles. START: /david/test/file3d
processing /david/test/file3d
listFiles. START: /david/test/file3d/file1
processing /david/test/file3d/file1
listFiles. START: /david/test/file3d/file2
processing /david/test/file3d/file2
listFiles. START: /david/test/file4
processing /david/test/file4