CODEWITHSUNDEEP
scalastreamakka
a-herch
a-herch

Reputation: 137

akka streams onComplete{} do not end

I'am trying to read multiples files with akka streams. Process the result and export processing results into files. program gets stuck in onComplete{}. Results are not exported untill i stop the program manually.

I tried the following code. exporting is not executed untill i stop the program. any other processing before or after the export gets executed any way even before stopping the program.

// 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 })  
}

}

Upvotes: 0

Views: 292

Answers (1)

vkt
vkt

Reputation: 1459

If you are running this as a script, you have to stop/terminate the Actor system or exist system manually.

Upvotes: 2

Related Questions