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.javadsl.Framing
import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source}
import akka.util.ByteString
import akka.{Done, NotUsed}
import com.typesafe.scalalogging.slf4j.Logger
import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException
import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.separator
import fr.natan.akkastreamfileprocessingapi.models.{Person, TvSeries}
import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists

import java.io.File
import java.nio.file.Paths
import scala.concurrent.Future

object AkkaStreamComponents {

  implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")

  private def convertToTvSerie(map: Map[String, String]): TvSeries = {
    val tvSerie: TvSeries = TvSeries(
      map("tconst"),
      map("titleType"),
      map("primaryTitle"),
      map("originalTitle"),
      map("isAdult"),
      map("startYear"),
      map("endYear"),
      map("runtimeMinutes"),
      map("genres")
    )

    tvSerie
  }

  private def convertToPerson(map: Map[String, String]): Person ={
    Person(
      map("nconst"),
      map("primaryName"),
      map("birthYear"),
      map("deathYear"),
      map("primaryProfession").split(",").toList,
      map("knownForTitles").split(",").toList,
    )
  }

  //flows

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

    val tvFlow: Flow[Map[String, String], TvSeries, NotUsed] =
      Flow[Map[String, String]]
        .map(row => {
          convertToTvSerie(row)
        })
    tvFlow
  }

  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]) => {
          convertToPerson(rowMap)
        })

    personFlow
  }

  def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: 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("primaryTitle","")==moviePrimaryTitle
        })
        .map(rowMap => {
          convertToTvSerie(map = rowMap)
        })

    filterFlow
  }

  def filterByPersonID(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("nconst","")==nconst
        })
        .map(rowMap=>{
          convertToPerson(map = rowMap)
        })

    personFilter
  }

  //source
  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))
      .via(CsvToMap.toMapAsStrings())

    datasource
  }

  def buildAndValidateSource(inputFile: File): Source[Map[String, String], NotUsed] = {

    val source: Source[Map[String, String], NotUsed] = buildSource(inputFile = inputFile)
    if (source == null) {
      throw new FileNotFoundException(filename = inputFile.getPath)
    }
    source
  }

  //sink
  def buildTvSeriesSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
    val tvSeriesSink : Sink[TvSeries, Future[Done]]= Sink.foreach[TvSeries](
      (movie: TvSeries) => {
        logger.info(s"${movie.toString}")
      }
    )
    tvSeriesSink
  }

  def buildPersonsSink(logger: Logger): Sink[Person,Future[Done]] = {
    val listPersonsSink: Sink[Person, Future[Done]]=
      Sink.foreach[Person]((person: Person)=>{
        logger.info(s"${person.toString}")
      })

    listPersonsSink
  }

  def buildPersonSink(logger: Logger): Sink[Person, Future[Done]] = {
    Sink.foreach[Person](
      (person: Person) => logger.info(s"${person.toString}")
    )
  }
}