Потоки Akka: чтение нескольких файлов

Приведенный ниже код не так краток, как мог бы быть, чтобы четко разделить различные проблемы.

// Given a stream of bytestrings delimited by the system line separator we can get lines represented as Strings
val lines = Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true).map(bs => bs.utf8String)

// given as stream of Paths we read those files and count the number of lines
val lineCounter = Flow[Path].flatMapConcat(path => FileIO.fromPath(path).via(lines)).fold(0l)((count, line) => count + 1).toMat(Sink.head)(Keep.right)

// Here's our test data source (replace paths with real paths)
val testFiles = Source(List("somePathToFile1", "somePathToFile2").map(new File(_).toPath))

// Runs the line counter over the test files, returns a Future, which contains the number of lines, which we then print out to the console when it completes
testFiles runWith lineCounter foreach println

Обновить. Я не увидел принятого ответа, потому что не обновил страницу> _ ‹. Я все равно оставлю это здесь, так как я также добавил несколько примечаний об обработке ошибок.

Я считаю, что следующая программа делает то, что вы хотите:

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, IOResult}
import akka.stream.scaladsl.{FileIO, Flow, Framing, Keep, Sink, Source}
import akka.util.ByteString
import scala.concurrent.{Await, Future}
import scala.util.{Failure, Success}
import scala.util.control.NonFatal
import java.nio.file.Paths
import scala.concurrent.duration._

object TestMain extends App {
  implicit val actorSystem = ActorSystem("test")
  implicit val materializer = ActorMaterializer()
  implicit def ec = actorSystem.dispatcher

  val sources = Vector("build.sbt", ".gitignore")
    .map(Paths.get(_))
    .map(p =>
      FileIO.fromPath(p)
        .viaMat(Framing.delimiter(ByteString(System.lineSeparator()), Int.MaxValue, allowTruncation = true))(Keep.left)
        .mapMaterializedValue { f =>
          f.onComplete {
            case Success(r) if r.wasSuccessful => println(s"Read ${r.count} bytes from $p")
            case Success(r) => println(s"Something went wrong when reading $p: ${r.getError}")
            case Failure(NonFatal(e)) => println(s"Something went wrong when reading $p: $e")
          }
          NotUsed
        }
    )
  val finalSource = Source(sources).flatMapConcat(identity)

  val result = finalSource.map(_ => 1).runWith(Sink.reduce[Int](_ + _))
  result.onComplete {
    case Success(n) => println(s"Read $n lines total")
    case Failure(e) => println(s"Reading failed: $e")
  }
  Await.ready(result, 10.seconds)

  actorSystem.terminate()
}

Ключевым моментом здесь является _ 2_: он преобразует каждый элемент потока в источник и возвращает поток элементов, полученных из этих источников, если они выполняются последовательно.

Что касается обработки ошибок, вы можете либо добавить обработчик future в аргументе mapMaterializedValue, либо обработать последнюю ошибку текущего потока, поместив обработчик на Sink.foreach материализованное будущее значение. Я сделал и то, и другое в приведенном выше примере, и если вы протестируете это, скажем, на несуществующем файле, вы увидите, что одно и то же сообщение об ошибке будет напечатано дважды. К сожалению, flatMapConcat() не собирает материализованные значения, и, честно говоря, я не вижу, как он мог бы это сделать разумно, поэтому при необходимости вы должны обрабатывать их отдельно.

У меня есть один неожиданный ответ - не используйте akka.FileIO. Кажется, это нормально работает, например:

val sources = Seq("sample.txt", "sample2.txt").map(io.Source.fromFile(_).getLines()).reduce(_ ++ _)
val source = Source.fromIterator[String](() => sources)
val lineCount = source.map(_ => 1).runWith(Sink.reduce[Int](_ + _))

Я все еще хотел бы знать, есть ли лучшее решение.