package fr.natan.akkastreamfileprocessingapi.service import akka.actor.ActorSystem import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap} import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source} import akka.{Done, NotUsed} import com.typesafe.scalalogging.slf4j.Logger import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSerie} import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.{buildPersonModel, buildTvSerieModel} import java.io.File import java.nio.file.Paths import scala.concurrent.Future //noinspection SpellCheckingInspection object AkkaStreamComponents { implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor") //flows building def buildPersonFlow(): Flow[Map[String, String], Person, NotUsed] = { val personFlow: Flow[Map[String, String], Person, NotUsed] = Flow[Map[String, String]] .map((rowMap: Map[String, String]) => { buildPersonModel(rowMap) }) personFlow } def filterPersonByIdFlow(personID: String): Flow[Map[String, String], Person, NotUsed]={ val personFlowFilter: Flow[Map[String, String], Person, NotUsed]= Flow[Map[String, String]] .filter((rowMap:Map[String, String])=>{ rowMap.getOrElse(key="nconst",default="")==personID }) .map(rowMap=>{ buildPersonModel(personMap = rowMap) }) personFlowFilter } def filterPersonByNameFlow(primaryName: String): Flow[Map[String, String], Person, NotUsed] ={ val personFlowFilter: Flow[Map[String, String], Person, NotUsed] = Flow[Map[String, String]] .filter((rowMap: Map[String, String]) =>{ rowMap.getOrElse(key = "primaryName", default = "")==primaryName }) .map((rowMap: Map[String, String])=>{ buildPersonModel(personMap = rowMap) }) personFlowFilter } def buildTvSerieFlow(): Flow[Map[String, String], TvSerie, NotUsed] = { val tvSerieFlow: Flow[Map[String, String], TvSerie, NotUsed] = Flow[Map[String, String]] .map((rowMap: Map[String, String]) => { buildTvSerieModel(tvSerieMap = rowMap) }) tvSerieFlow } def filterTvSerieByIdFlow(tvSerieID: String): Flow[Map[String, String], TvSerie, NotUsed] = { val tvSerieFlowFilter : Flow[Map[String, String], TvSerie, NotUsed] = Flow[Map[String, String]] .filter((rowMap: Map[String, String])=>rowMap.getOrElse(key = "tconst", default = "")==tvSerieID) .map((rowMap: Map[String, String])=>{ buildTvSerieModel(tvSerieMap = rowMap) }) tvSerieFlowFilter } def filterTvSerieByPrimaryTitleFlow(tvSeriePrimaryTitle: String): Flow[Map[String, String], TvSerie, NotUsed] = { val filterFlowFilter: Flow[Map[String, String], TvSerie, NotUsed] = Flow[Map[String, String]] .filter((rows: Map[String, String]) => { rows.getOrElse(key = "primaryTitle", default = "") == tvSeriePrimaryTitle }) .map(rowMap => { val tvSerie: TvSerie = buildTvSerieModel(tvSerieMap = rowMap) tvSerie }) filterFlowFilter } //source building def buildSource(inputFile: File): Source[Map[String, String], NotUsed] = { var datasource: Source[Map[String, String], NotUsed] = null if (!fileExists(inputFile.getPath)) { return null } datasource = Source .single(inputFile) .flatMapConcat((filename: File) => { FileIO.fromPath( Paths.get(filename.getPath) ) } ) .via(Compression.gunzip()) .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote)) .via(CsvToMap.toMapAsStrings()) datasource } def buildAndValidateSource(inputFile: File): Source[Map[String, String], _] = { val source: Source[Map[String, String], _] = buildSource(inputFile = inputFile) if (source == null) { throw new FileNotFoundException(filename = inputFile.getPath) } source } //sinks building def buildAllTvSeriesSink(logger: Logger): Sink[TvSerie, Future[Done]] = { val tvSeriesSink: Sink[TvSerie, Future[Done]] = Sink .foreach((tvSerie: TvSerie)=>logger.info(s"${tvSerie.toString}")) tvSeriesSink } def buildAllPersonsSink(logger: Logger): Sink[Person,Future[Done]] = { val listPersonsSink: Sink[Person, Future[Done]]= Sink.foreach((person: Person)=>logger.info(s"${person.toString}")) listPersonsSink } }