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
import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
  buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildSource, buildTvSerieFlow,
  filterByPersonIdFlow, filterByPersonNameFlow, filterByTvSeriePrimaryTitleFlow
}
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component

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

//noinspection SpellCheckingInspection
@Component
class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {

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

  override def getPersonById(nconst: String): Future[Person] = {
    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
    val res = source
      .via(flow = filterByPersonIdFlow(nconst = nconst))
      .runWith(Sink.head)

    res.andThen {
      case Failure(exception) => logger.info(s"$exception")
      case Success(value) => logger.info(s"$value")
    }

    res
  }

  override def getPersonByName(primaryName: String): Future[IndexedSeq[Person]] = {
    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
    val persons: Future[IndexedSeq[Person]] = source
      .via(flow = filterByPersonNameFlow(primaryName = primaryName))
      .runWith(Sink.collection)

    persons.onComplete({
      case Failure(exception) => logger.error(s"$exception")
      case Success(value: IndexedSeq[Person]) =>
        value.foreach((person: Person)=>logger.info(s"$person"))
        logger.info("SUCCESS")
    })

    persons
  }

  override def getTvSerieByPrimaryTitle(tvSeriePrimaryTitle: String):Future[IndexedSeq[TvSeries]] = {

    val tvSeriesSource: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)

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

    val tvSries :Future[IndexedSeq[TvSeries]] = tvSeriesSource
      .via(flow = filterByTvSerieTitle)
      .runWith(Sink.collection)

      tvSries.onComplete({
      case Failure(exception) => logger.info(s"$exception")
      case Success(value: IndexedSeq[TvSeries]) =>
        value.foreach((tvSrie: TvSeries)=>logger.info(s"$tvSrie"))
        logger.info("SUCCESS")
      })

    tvSries
  }

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

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

    res
  }

  private def getListOfPersonsIDByTvSerieID(tvSerieID: Future[Option[String]]): Future[IndexedSeq[Option[String]]]={
    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
    val res: Future[IndexedSeq[Option[String]]] = source
      .filter((rowMaps: Map[String, String])=>{
        rowMaps.getOrElse(key = "tconst", default = "")==tvSerieID.value.get.get.get
      })
      .map((rowMap: Map[String, String])=>{
        rowMap.get(key = "nconst")
      })
      .runWith(Sink.collection)

    res
  }

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

    val res : Future[IndexedSeq[Person]]=
      source
        .filter((rowMaps: Map[String, String])=>{
          listPersonsIDs.value.get.get.contains(rowMaps.get(key = "nconst"))
        })
      .map((rowMap: Map[String, String])=>{
       buildPersonModel(rowMap)
      })
      .runWith(Sink.collection)

    res
  }
  override def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String):Future[IndexedSeq[Person]]={

    logger.info("STEP 1/3")
    val tvSerieID : Future[Option[String]]= getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
    tvSerieID.andThen({
      case Failure(exception) => logger.error(s"!!!!!!!!!!!!!!!!!!!!1$exception")
      case Success(value: Option[String]) =>
        logger.info(s"TvSerie ID: $value")
        logger.info("END STEP 1/3")
    }).flatMap({
      future =>
        logger.info("STEP 2/3")
        val listPersonIDs: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieID = tvSerieID)
        listPersonIDs.andThen({
          case Failure(exception) => logger.error(s"!!!!!!!!!!!!!!!!!!!!!2$exception")
          case Success(value) =>
            value.foreach((personID: Option[String]) => logger.info(s"Person ID:$personID"))
            logger.info("END STEP 2/3")
        }).flatMap({
          future =>
            logger.info("STEP 3/3")
            val personsTeam: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDs = listPersonIDs)
            personsTeam.andThen({
              case Failure(exception) => logger.error(s"!!!!!!!!!!!!!!!!!!!!!!!3$exception")
              case Success(value) =>
                logger.info(s"$value")
                value.foreach((person: Person) => logger.info(s"${person.toString}"))
                logger.info("END STEP 3/3")
            })
        })
    })

   null
  }

  override def getAllPersons(): Unit = {
    val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
    //graph
    val startTime: Long = System.currentTimeMillis()
    val result: Future[Done] = personSource
      .via(flow = buildPersonFlow())
      .runWith(sink = buildAllPersonsSink(logger = logger))

    result.onComplete {
      case Failure(exception) => logger.error(s"$exception")
      case Success(value) =>
        logger.info(s"$value")
        val time: Long = (System.currentTimeMillis() - startTime) / 100
        logger.info(s"elapsed time: $time")
    }
  }

  override def getAllTvSeries(): Unit = {
    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
    val sink: Sink[TvSeries, Future[Done]] = buildAllTvSeriesSink(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")
      }
  }

}