akka-stream-with-springboot-api/src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamComponents.scala

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, TvSeries}
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 filterByTvSeriePrimaryTitleFlow(tvSeriePrimaryTitle: String): Flow[Map[String, String], TvSeries, NotUsed] = {
    val filterFlow: Flow[Map[String, String], TvSeries, NotUsed] = Flow[Map[String, String]]
        .filter((rows: Map[String, String]) => {
          rows.getOrElse(key = "primaryTitle", default = "")==tvSeriePrimaryTitle
        })
        .map(rowMap => {
          val tvSerie: TvSeries = buildTvSerieModel(tvSerieMap = rowMap)
          tvSerie
        })

    filterFlow
  }

  def filterByPersonIdFlow(nconst: String): Flow[Map[String, String], Person, NotUsed]={
    val personFilter: Flow[Map[String, String], Person, NotUsed]=
      Flow[Map[String, String]]
        .filter((rowMap:Map[String, String])=>{
          rowMap.getOrElse(key="nconst",default="")==nconst
        })
        .map(rowMap=>{
          buildPersonModel(personMap = rowMap)
        })

    personFilter
  }

  def filterByPersonNameFlow(primaryName: String): Flow[Map[String, String], Person, NotUsed] ={
    val personFilter: 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)
        })

    personFilter
  }

  def buildTvSerieFlow(): Flow[Map[String, String], TvSeries, NotUsed] = {

    val tvFlow: Flow[Map[String, String], TvSeries, NotUsed] =
      Flow[Map[String, String]]
        .map((rowMap: Map[String, String]) => {
          buildTvSerieModel(tvSerieMap = rowMap)
        })
    tvFlow
  }
  //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[TvSeries, Future[Done]] = {
    val tvSeriesSink: Sink[TvSeries, Future[Done]] = Sink
      .foreach((tvSerie: TvSeries)=>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
  }
}