Я пытаюсь прочитать несколько файлов с потоками Акка. Обработайте результат и экспортируйте результаты обработки в файлы.
программа застревает в onComplete {}. Результаты не экспортируются, пока я не остановлю программу вручную.
Я попробовал следующий код. экспорт не выполняется, пока я не остановлю программу. любая другая обработка до или после экспорта выполняется любым способом, даже до остановки программы.
// this function reads multiples files
@throws[FileNotFoundException]
def concatFilesAkka(path : String, date : String, numberOfDays :
Int) : Future[Seq[String]] = {
implicit val system = ActorSystem("Sys")
val settings = ActorMaterializerSettings(system)
implicit val materializer = ActorMaterializer(settings)
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
val formattedDate = LocalDate.parse(date, formatter);
def files = {.. }
val result = Source(files).flatMapConcat(filename =>
FileIO.fromPath(Paths.get(filename))
.via(Framing.delimiter(ByteString("\n"), 256, allowTruncation =
true).map(_.utf8String))
).toMat(Sink.seq)(Keep.right)
.run()
result
}
// processing
def process(date: String, inPath: String, outPath: String,
..): Unit = {
implicit val system = ActorSystem("Sys")
val settings = ActorMaterializerSettings(system)
implicit val materializer = ActorMaterializer(settings)
val a = concatFilesAkka(inPath,date, numberOfDays)
a.onComplete(x => {
var transactions = x.get.map(line => line.split('|')).toList
val groupedByMagasin = transactions.filter(x => x(3) !=
"0").groupBy(x
=> x(2)).foreach(x => {
.......
.......
val top100VenteGlobale: List[(String, Int)] = Nil
val top100CaGlobale: List[(String, Int)] = Nil
...
...
})
//exporting top 100 vente global et top 100 ca global
export(top100VenteGlobale, outPath + "top_100_vente_GLOBAL_" +
date })
}
}