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

package fr.natan.akkastreamfileprocessingapi.service

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.{Done, NotUsed}
import com.typesafe.scalalogging.slf4j.Logger
import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.{buildPersonModel, buildTvSerieModel}
import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
  buildAllPersonsSink,
  buildAndValidateSource,
  buildPersonFlow,
  buildPersonSink,
  buildSource,
  buildTvSerieFlow,
  buildTvSeriesSink,
  filterByPersonIdFlow,
  filterByPersonNameFlow,
  filterByTvSeriePrimaryTitleFlow
}
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.DurationInt
import scala.concurrent.{Await, Future}
import scala.language.postfixOps
import scala.util.{Failure, Success}

@Component
class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {

  implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
  implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))

  override def getAllPersons(): Unit = {
    val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
    val personSink: Sink[Person, Future[Done]] = buildAllPersonsSink(logger = logger)

    //graph
    val startTime: Long = System.currentTimeMillis()
    val result: Future[Done] = personSource
      .via(flow = buildPersonFlow())
      .runWith(sink = personSink)
      .andThen {
        case Success(value) =>
          val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
          logger.info(s"$value: Successfully processed, elapsed time: $elapsedTime")
        case Failure(exception) => logger.error(s"$exception: Fail")
      }
  }

  override def getPersonById(nconst: String): Person = {
    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)

    val startTime: Long = System.currentTimeMillis()
    val res = source
      .via(flow = filterByPersonIdFlow(nconst = nconst))
      .runWith(sink = buildPersonSink())
     .andThen {
        case Success(value) =>
          val elapsedTime: Long = (System.currentTimeMillis()-startTime)/1000
          logger.info(s"$value")
          logger.info(s"Successfully processed, elapsed time: $elapsedTime")
        case Failure(exception) => logger.error(s"$exception: Fail")
      }

    Await.result(res,1 minutes)
    val person: Person = res.value.get.get
    person
  }

  override def getPersonByName(primaryName: String): Unit = {
    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
    source
      .via(
        flow = filterByPersonNameFlow(personName = primaryName)
      )
      .runWith(sink = buildPersonSink())
      .andThen {
        case Failure(exception) => logger.info(s"$exception")
        case Success(value) => logger.info(s"$value: Successufully processing")
      }
  }

  override def getAllTvSeries(): Unit = {

    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
    val sink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)

    val startingTime: Long = System.currentTimeMillis()

    //graph sink->flow->sink
    source
      .via(flow = buildTvSerieFlow())
      .runWith(sink = sink)
      .andThen {
        case Success(value) =>
          val elapsedTime: Long = (System.currentTimeMillis() - startingTime) / 1000
          logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
        case Failure(error: Error) => logger.error(s"$error")
      }

  }

  override def getByTvSeriePrimaryTitle(tvSeriePrimaryTitle: String): Unit = {

    val tvSeriesSource: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
    val tvSeriesSink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)

    val filterByMovieTitleFlow: Flow[Map[String, String], TvSeries, NotUsed] =
      filterByTvSeriePrimaryTitleFlow(tvSeriePrimaryTitle = tvSeriePrimaryTitle)

    val startTime: Long = System.currentTimeMillis()
    val listTvSeries: Future[Done] = tvSeriesSource
      .via(flow = filterByMovieTitleFlow)
      .runWith(sink = tvSeriesSink)
      .andThen {
        case Success(value) =>
          val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
          logger.info(s"$value: successfully processing file, elapsed time: $elapsedTime sec")
        case Failure(error: Error) => logger.error(s"$error")
      }

    getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)

  }

  private def getTvSerieIdByPrimaryTitle(primaryTitle: String): String = {
    val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)

    var tvSerie: TvSeries = null
    val res: Future[Option[String]] = source
      .filter((rowMap: Map[String, String]) => {
        rowMap.getOrElse(key = "primaryTitle", "") == primaryTitle
      })
      .map((rowMap: Map[String, String]) => {
        tvSerie = buildTvSerieModel(rowMap)
        rowMap.get("tconst")
      })
      .runWith(Sink.head)

    Await.result(res, 1 minutes)
    val tconst: String = res.value.get.get.get
    logger.info(s"TvSerie ID: $tconst, $tvSerie")
    tconst
  }

  private def getListOfPersonsIDByTvSerieID(tconst: String): List[Option[String]]={
    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)

    val startTime : Long = System.currentTimeMillis()
    val res: Future[IndexedSeq[Option[String]]] = source
      .filter((rowMaps: Map[String, String])=>{
        rowMaps.getOrElse(key = "tconst", default = "")==tconst
      })
      .map((rowMap: Map[String, String])=>{
        rowMap.get(key = "nconst")
      })
      .runWith(Sink.collection)
      .andThen {
        case Success(value) =>
          logger.info(s"$value")
          val elapsedTime : Long = (System.currentTimeMillis() - startTime)/1000
          logger.info(s"success, elapsed time: $elapsedTime")

        case Failure(error: Error) => logger.error(s"$error")
      }

    Await.result(res, 3 minutes)
    val listPersonsID: List[Option[String]] = res.value.get.get.toList

    listPersonsID

  }

  private def getListOfPersonsForTvSerie(nconstList: List[Option[String]]): List[Person] = {
    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)

    val startTime: Long = System.currentTimeMillis()

    val res : Future[IndexedSeq[Person]]= source
      .filter((rowMap: Map[String, String])=>{
        nconstList.contains(rowMap.getOrElse(key = "nconst",default = ""))
      })
      .map((rowMap: Map[String, String])=>{
        val person: Person = buildPersonModel(rowMap)
        person
      })
      .runWith(Sink.collection)
      .andThen {
        case Failure(exception) => logger.error(s"$exception")
        case Success(value) =>
          val elapsedTime: Long = (System.currentTimeMillis()-startTime)/1000
          logger.info(s"$value")
          logger.info(s"END: Successfully, elapsed time: $elapsedTime")
      }

    Await.result(res, 4 minutes)
    val listPersons: List[Person] = res.value.get.get.toList

    listPersons

  }

  override def generalize(tvSeriePrimaryTitle: String): Unit={

    val tvSerieID: String = getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
    val listPersonIDs: List[Option[String]] = getListOfPersonsIDByTvSerieID(tconst = tvSerieID)
    getListOfPersonsForTvSerie(nconstList = listPersonIDs)
  }

}