first commit

src/main/java/fr/natan/akkastreamfileprocessingapi/controller/TvSeriesController.java

@@ -13,13 +13,11 @@ import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
 import play.api.libs.json.JsValue;
 import play.api.libs.json.Json;
-import scala.Option;
 import scala.collection.IndexedSeq;
 import scala.concurrent.Future;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
@@ -30,7 +28,6 @@ public class TvSeriesController {
 
     private final AkkaStreamFileProcessingFuture akkaStreamFilesProcessing;
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
-    private static final int BORN_INF = 100, BORN_SUP = 2000;
 
     public TvSeriesController(AkkaStreamFileProcessingFuture akkaStreamFilesProcessing) {
         this.akkaStreamFilesProcessing = akkaStreamFilesProcessing;
@@ -38,13 +35,9 @@ public class TvSeriesController {
 
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
     private String getPersonByID(@PathVariable(name = "personID") String personID) throws ExecutionException, InterruptedException {
-        Future<Option<ModelsAndJsonMap.Person>> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
+        Future<ModelsAndJsonMap.Person> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
         CompletableFuture<ModelsAndJsonMap.Person> completableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(futurePerson);
-        while (!completableFuture.isDone()) {
-            logger.info("IS PROCESSING...");
-            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
-        }
         ModelsAndJsonMap.Person person = completableFuture.get();
 
         JsValue personJs = Json.toJson(person, person.personJsonFormatter());
@@ -55,13 +48,9 @@ public class TvSeriesController {
     private String getPersonByName(@PathVariable(name = "primaryName") String primaryName)
             throws ExecutionException, InterruptedException {
 
-        Future<Option<ModelsAndJsonMap.Person>> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
+        Future<ModelsAndJsonMap.Person> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
         CompletableFuture<ModelsAndJsonMap.Person> personCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(personFuture);
-        while (!personCompletableFuture.isDone()) {
-            logger.info("IS PROCESSING...");
-            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
-        }
         ModelsAndJsonMap.Person person = personCompletableFuture.get();
         JsValue personJs = Json.toJson(person, person.personJsonFormatter());
         return Json.prettyPrint(personJs);
@@ -70,13 +59,10 @@ public class TvSeriesController {
     @RequestMapping(value = "/tvseries/id/{tvSerieId}", method = RequestMethod.GET)
     private String getTvSerieByID(@PathVariable(name = "tvSerieId") String tvSerieID)
             throws ExecutionException, InterruptedException {
-        Future<Option<ModelsAndJsonMap.TvSerie>> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
+        Future<ModelsAndJsonMap.TvSerie> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
         CompletableFuture<ModelsAndJsonMap.TvSerie> tvSerieCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(tvSerieFuture);
-        while (!tvSerieCompletableFuture.isDone()) {
-            logger.info("IS PROCESSING...");
-            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
-        }
+
         ModelsAndJsonMap.TvSerie tvSerie = tvSerieCompletableFuture.get();
         JsValue tvSerieJs = Json.toJson(tvSerie, tvSerie.tvSerieJsonFormatter());
         return Json.prettyPrint(tvSerieJs);
@@ -90,10 +76,7 @@ public class TvSeriesController {
                 .getTvSeriesByPrimaryTitleFuture(tvseriePrimaryTitle);
         CompletableFuture<IndexedSeq<ModelsAndJsonMap.TvSerie>> completableFuture =
                 CompletableFutureBuilder.buildcompletableFuture2(listFuture);
-        while (!completableFuture.isDone()) {
-            logger.info("IS PROCESSING...");
-            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
-        }
+
         IndexedSeq<ModelsAndJsonMap.TvSerie> tvSerieList = completableFuture.get();
         List<String> tvSeries = new ArrayList<>();
         tvSerieList.foreach(tvSerie -> {
@@ -113,10 +96,6 @@ public class TvSeriesController {
                 .getPersonsForTvSerieByTvSerieTitleFuture(tvSerieTitle);
         CompletableFuture<IndexedSeq<ModelsAndJsonMap.Person>> completableFuture = CompletableFutureBuilder
                 .buildcompletableFuture2(futurePersonSeq);
-        while (!completableFuture.isDone()){
-            logger.info("IS PROCESSING...");
-            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
-        }
 
         IndexedSeq<ModelsAndJsonMap.Person> personIndexedSeq = completableFuture.get();
         List<String> personList = new ArrayList<>();
@@ -138,10 +117,6 @@ public class TvSeriesController {
         CompletableFuture<IndexedSeq<ModelsAndJsonMap.Person>> seqCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture2(futureIndexedSeqPersons);
 
-        while (!seqCompletableFuture.isDone()){
-            logger.info("IS PROCESSING...");
-            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
-        }
         IndexedSeq<ModelsAndJsonMap.Person> personIndexedSeq = seqCompletableFuture.get();
         List<String> personsList = new ArrayList<>();
         personIndexedSeq.foreach(person ->{

src/main/java/fr/natan/akkastreamfileprocessingapi/exceptionshandler/BusinessExceptionHandler.java

@@ -1,7 +1,7 @@
 package fr.natan.akkastreamfileprocessingapi.exceptionshandler;
 
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException;
-import fr.natan.akkastreamfileprocessingapi.businessexceptions.MovieNotFoundException;
+import fr.natan.akkastreamfileprocessingapi.businessexceptions.TvSerieNotFoundException;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.ControllerAdvice;
@@ -15,8 +15,9 @@ public class BusinessExceptionHandler {
         return new ResponseEntity<>("The File Is Not Found Exception", HttpStatus.NOT_FOUND);
     }
 
-    @ExceptionHandler(value = MovieNotFoundException.class)
-    private ResponseEntity<String> handleMovieNotFoundException(){
+    @ExceptionHandler(value = TvSerieNotFoundException.class)
+    private ResponseEntity<String> handleTvSerieNotFoundException(){
         return new ResponseEntity<>("Movie Title Not Found Exception", HttpStatus.NOT_FOUND);
     }
+
 }

src/main/java/fr/natan/akkastreamfileprocessingapi/futurecompleteness/CompletableFutureBuilder.java

@@ -1,25 +1,31 @@
 package fr.natan.akkastreamfileprocessingapi.futurecompleteness;
 
-import scala.Option;
 import scala.collection.IndexedSeq;
 import scala.concurrent.Future;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
 
 
 public class CompletableFutureBuilder<T> {
 
-    public static<T> CompletableFuture<T> buildcompletableFuture1(Future<Option<T>> futureT) {
+   private static final Logger logger = Logger.getLogger("log");
+    public static<T> CompletableFuture<T> buildcompletableFuture1(Future<T> futureT) {
         CompletableFuture<T> completableFuture = new CompletableFuture<>();
+
+
         Executors.newSingleThreadExecutor().submit(() -> {
             while(!futureT.isCompleted()){
-                TimeUnit.MILLISECONDS.sleep(300);
+                try {
+                    TimeUnit.MILLISECONDS.sleep(300);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
             }
-            T task = futureT.value().get().get().get();
+            T task = futureT.value().get().get();
             completableFuture.complete(task);
-            return null;
         });
 
         return completableFuture;
@@ -30,12 +36,16 @@ public class CompletableFutureBuilder<T> {
         CompletableFuture<IndexedSeq<T>> completableFuture = new CompletableFuture<>();
         Executors.newSingleThreadExecutor().submit(() -> {
             while (!futureListT.isCompleted()) {
-                TimeUnit.MILLISECONDS.sleep(300);
+                try {
+                    TimeUnit.MILLISECONDS.sleep(300);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
             }
 
             IndexedSeq<T> task = futureListT.value().get().get();
             completableFuture.complete(task);
-            return null;
+
         });
 
         return completableFuture;

src/main/scala/fr/natan/akkastreamfileprocessingapi/businessexceptions/MovieNotFoundException.scala → src/main/scala/fr/natan/akkastreamfileprocessingapi/businessexceptions/PersonNotFoundException.scala

@@ -1,3 +1,3 @@
 package fr.natan.akkastreamfileprocessingapi.businessexceptions
 
-class MovieNotFoundException extends Exception{}
+class PersonNotFoundException extends Exception{}

src/main/scala/fr/natan/akkastreamfileprocessingapi/businessexceptions/TvSerieNotFoundException.scala

@@ -0,0 +1,3 @@
+package fr.natan.akkastreamfileprocessingapi.businessexceptions
+
+class TvSerieNotFoundException extends Exception{}

src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamComponents.scala

@@ -7,8 +7,8 @@ import akka.{Done, NotUsed}
 import com.typesafe.scalalogging.slf4j.Logger
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException
 import fr.natan.akkastreamfileprocessingapi.models.ModelsAndJsonMap.{Person, TvSerie}
-import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists
 import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.{buildPersonModel, buildTvSerieModel}
+import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists
 
 import java.io.File
 import java.nio.file.Paths
@@ -91,7 +91,7 @@ object AkkaStreamComponents {
 
   //source building
 
-  def buildSource(inputFile: File): Source[Map[String, String], NotUsed] = {
+  private def buildSource(inputFile: File): Source[Map[String, String], NotUsed] = {
 
     var datasource: Source[Map[String, String], NotUsed] = null
     if (!fileExists(inputFile.getPath)) {

src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamFileProcessingFuture.scala

@@ -5,12 +5,11 @@ import fr.natan.akkastreamfileprocessingapi.models.ModelsAndJsonMap.{Person, TvS
 
 import scala.concurrent.Future
 
-//noinspection SpellCheckingInspection,AccessorLikeMethodIsUnit
 trait AkkaStreamFileProcessingFuture {
-  def getPersonByIdFuture(personID: String): Future[Option[Person]]
-  def getPersonByNameFuture(primaryName: String):  Future[Option[Person]]
+  def getPersonByIdFuture(personID: String): Future[Person]
+  def getPersonByNameFuture(primaryName: String):  Future[Person]
 
-  def getTvSerieByIdFuture(tvSerieID: String): Future[Option[TvSerie]]
+  def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie]
   def getTvSeriesByPrimaryTitleFuture(tvSerieTitle: String):Future[IndexedSeq[TvSerie]]
 
   def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]

src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamFileProcessingImpl.scala

@@ -3,13 +3,13 @@ package fr.natan.akkastreamfileprocessingapi.service
 import akka.Done
 import akka.stream.scaladsl.{Sink, Source}
 import com.typesafe.scalalogging.slf4j.Logger
-import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics}
+import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
 import fr.natan.akkastreamfileprocessingapi.models.ModelsAndJsonMap.{Person, TvSerie}
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
-  actorSystem, buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildSource,
-  buildTvSerieFlow, filterPersonByIdFlow, filterPersonByNameFlow, filterTvSerieByIdFlow, filterTvSerieByPrimaryTitleFlow}
-import fr.natan.akkastreamfileprocessingapi.service.UtilitiesClass.{
-  getListOfPersonsForTvSerie, getListOfPersonsIDByTvSerieID, getTvSerieIDFuture, getTvSerieIdByPrimaryTitle}
+  actorSystem, buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildTvSerieFlow,
+  filterPersonByIdFlow, filterPersonByNameFlow, filterTvSerieByIdFlow, filterTvSerieByPrimaryTitleFlow}
+import fr.natan.akkastreamfileprocessingapi.service.UtilitiesClass.{getListOfPersonsForTvSerie, getListOfPersonsIDByTvSerieID,
+  getTvSerieIDFuture, getTvSerieIdByPrimaryTitle}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -24,46 +24,43 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
 
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getPersonByIdFuture(personID: String): Future[Option[Person]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+  override def getPersonByIdFuture(personID: String): Future[Person] = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = nameBasics)
 
     val start: Long = System.currentTimeMillis()
-    val personFuture: Future[Option[Person]] = source
+    val personFuture: Future[Person] = source
       .via(flow = filterPersonByIdFlow(personID = personID))
-      .runWith(Sink.headOption[Person])
+      .runWith(Sink.head[Person])
 
     personFuture.andThen({
-      case Failure(exception) => logger.info(s"$exception")
-      case Success(value: Option[Person]) =>
-        logger.info(s"$value")
-        logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
+      case Failure(exception) => logger.error(s"!${exception.printStackTrace()}")
     })
 
     personFuture
   }
 
-  override def getPersonByNameFuture(primaryName: String): Future[Option[Person]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+  override def getPersonByNameFuture(primaryName: String): Future[Person] = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = nameBasics)
 
     val start: Long = System.currentTimeMillis()
-    val personFuture: Future[Option[Person]] = source
+    val personFuture: Future[Person] = source
       .via(flow = filterPersonByNameFlow(primaryName = primaryName))
-      .runWith(Sink.headOption[Person])
+      .runWith(Sink.head[Person])
 
     personFuture
   }
 
-  override def getTvSerieByIdFuture(tvSerieID: String): Future[Option[TvSerie]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
+  override def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie] = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
 
     val start: Long = System.currentTimeMillis()
-    val tvSerieFuture: Future[Option[TvSerie]] = source
+    val tvSerieFuture: Future[TvSerie] = source
       .via(flow = filterTvSerieByIdFlow(tvSerieID = tvSerieID))
-      .runWith(Sink.headOption[TvSerie])
+      .runWith(Sink.head[TvSerie])
 
     tvSerieFuture.onComplete({
       case Failure(exception) => logger.info(s"$exception")
-      case Success(value: Option[TvSerie]) =>
+      case Success(value: TvSerie) =>
         logger.info(s"$value")
         logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
     })
@@ -91,13 +88,15 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
   override def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]] = {
 
     //futures chaining
-    logger.info("STEP 1/3 START")
+    logger.info("STEP 1/3 ...")
     val start1: Long = System.currentTimeMillis()
-    val tvSerieIDFuture: Future[Option[String]] = getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
+    val tvSerieIDFuture: Future[Option[String]] = getTvSerieIdByPrimaryTitle(
+      primaryTitle = tvSeriePrimaryTitle,
+      inputFile = titleBasics
+    )
 
     val finalFuture: Future[IndexedSeq[Person]] =
       tvSerieIDFuture.andThen({
-        case Failure(exception) => logger.error(s"$exception")
         case Success(value: Option[String]) =>
           logger.info(s"TvSerie ID: $value")
           logger.info(s"STEP 1/3 END, elapsed time:${(System.currentTimeMillis() - start1) / 1000} sec")
@@ -105,24 +104,23 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
       })
         .flatMap({
           _ =>
-            logger.info("STEP 2/3 START")
+            logger.info("STEP 2/3 ...")
             val start2: Long = System.currentTimeMillis()
             val listPersonIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
-              tvSerieIdFuture = tvSerieIDFuture)
+              tvSerieIdFuture = tvSerieIDFuture, inputFile = titlePrincipalsBasics
+            )
             listPersonIDsFuture.andThen({
-              case Failure(exception) => logger.error(s"$exception")
               case Success(value) =>
                 value.foreach((personID: Option[String]) => logger.info(s"Person ID:$personID"))
                 logger.info(s"STEP 2/3 END, elapsed time:${(System.currentTimeMillis() - start2) / 1000} sec")
             })
               .flatMap({
                 _ =>
-                  logger.info("STEP 3/3 START")
+                  logger.info("STEP 3/3 ...")
                   val start3: Long = System.currentTimeMillis()
                   val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(
-                    listPersonsIDsFuture = listPersonIDsFuture)
+                    listPersonsIDsFuture = listPersonIDsFuture, inputFile = nameBasics)
                   personsTeamFuture.andThen({
-                    case Failure(exception) => logger.error(s"$exception")
                     case Success(value: IndexedSeq[Person]) =>
                       value.foreach((person: Person) => logger.info(s"${person.toString}"))
                       logger.info(s"STEP 3/3 END, elapsed time:${(System.currentTimeMillis() - start3) / 1000} sec")
@@ -136,8 +134,9 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
   override def getPersonsForTvSerieByTvSerieIDFuture(tvSerieId: String): Future[IndexedSeq[Person]] = {
 
     val start1: Long = System.currentTimeMillis()
-    logger.info("STEP 1/3")
-    val tvSerieIdFuture: Future[Option[String]] = getTvSerieIDFuture(tvSerieId = tvSerieId)
+    logger.info("STEP 1/3 ...")
+    val tvSerieIdFuture: Future[Option[String]] = getTvSerieIDFuture(
+      tvSerieID = tvSerieId, logger = logger, inputFile = titleBasics)
     tvSerieIdFuture.andThen({
       case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
       case Success(value: Option[String]) =>
@@ -147,10 +146,10 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     })
       .flatMap({
         _ =>
-          logger.info("STEP 2/3")
+          logger.info("STEP 2/3 ...")
           val start2: Long = System.currentTimeMillis()
           val listOfPersonsIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
-            tvSerieIdFuture = tvSerieIdFuture)
+            tvSerieIdFuture = tvSerieIdFuture, inputFile = titlePrincipalsBasics)
           listOfPersonsIDsFuture.andThen({
             case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
             case Success(value: IndexedSeq[Option[String]]) =>
@@ -159,10 +158,10 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
           })
             .flatMap({
               _ =>
-                logger.info("STEP 3/3")
+                logger.info("STEP 3/3 ...")
                 val start3: Long = System.currentTimeMillis()
                 val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(
-                  listPersonsIDsFuture = listOfPersonsIDsFuture)
+                  listPersonsIDsFuture = listOfPersonsIDsFuture, inputFile = nameBasics)
                 personsTeamFuture.andThen({
                   case Failure(exception) => logger.error(s"$exception")
                   case Success(value: IndexedSeq[Person]) =>
@@ -175,7 +174,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
   }
 
   override def getAllPersonsFuture: Future[Done] = {
-    val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+    val personSource: Source[Map[String, String], _] = buildAndValidateSource(inputFile = nameBasics)
     //graph
     val startTime: Long = System.currentTimeMillis()
     val result: Future[Done] = personSource
@@ -189,7 +188,6 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
         val time: Long = (System.currentTimeMillis() - startTime) / 100
         logger.info(s"elapsed time: $time")
     }
-
     result
   }
 
@@ -211,7 +209,6 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
 
     results
   }
-
 }
 
 

src/main/scala/fr/natan/akkastreamfileprocessingapi/service/UtilitiesClass.scala

@@ -1,17 +1,22 @@
 package fr.natan.akkastreamfileprocessingapi.service
 
-import akka.stream.scaladsl.{Sink, Source}
-import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
+import akka.NotUsed
+import akka.stream.scaladsl.{Flow, Sink, Source}
+import com.typesafe.scalalogging.slf4j.Logger
+import fr.natan.akkastreamfileprocessingapi.businessexceptions.TvSerieNotFoundException
 import fr.natan.akkastreamfileprocessingapi.models.ModelsAndJsonMap.Person
 import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildPersonModel
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{actorSystem, buildSource}
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{actorSystem, buildAndValidateSource}
 
+import java.io.File
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
+import scala.util.{Failure, Success}
 
 object UtilitiesClass {
 
-  def getTvSerieIdByPrimaryTitle(primaryTitle: String): Future[Option[String]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
+  def getTvSerieIdByPrimaryTitle(primaryTitle: String, inputFile: File): Future[Option[String]] = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = inputFile)
     val tvSerieIdFuture: Future[Option[String]] = source
       .filter((rowMap: Map[String, String]) => {
         rowMap.getOrElse(key = "primaryTitle", default = "") == primaryTitle
@@ -22,8 +27,8 @@ object UtilitiesClass {
     tvSerieIdFuture
   }
 
-  def getListOfPersonsIDByTvSerieID(tvSerieIdFuture: Future[Option[String]]): Future[IndexedSeq[Option[String]]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
+  def getListOfPersonsIDByTvSerieID(tvSerieIdFuture: Future[Option[String]], inputFile: File): Future[IndexedSeq[Option[String]]] = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = inputFile)
     val personsIDsFuture: Future[IndexedSeq[Option[String]]] = source
       .filter((rowMaps: Map[String, String]) => {
         rowMaps.getOrElse(key = "tconst", default = "") == tvSerieIdFuture.value.get.get.get
@@ -36,9 +41,9 @@ object UtilitiesClass {
     personsIDsFuture
   }
 
-  def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]]): Future[IndexedSeq[Person]] = {
+  def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]], inputFile: File): Future[IndexedSeq[Person]] = {
 
-    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = inputFile)
     val personsFuture: Future[IndexedSeq[Person]] = source
       .filter((rowMaps: Map[String, String]) => {
         listPersonsIDsFuture.value.get.get.contains(rowMaps.get(key = "nconst"))
@@ -51,14 +56,28 @@ object UtilitiesClass {
     personsFuture
   }
 
-  def getTvSerieIDFuture(tvSerieId: String): Future[Option[String]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
+  private def getTvSerieIdFlow(tvSerieID: String): Flow[Map[String, String], Option[String], NotUsed] = {
+    val tvSerieIDFlow = Flow[Map[String, String]]
+      .filter(rowMap => rowMap.getOrElse(key = "tconst", default = "") == tvSerieID)
+      .map(row => row.get("tconst"))
 
-    val tvSerieIdFuture: Future[Option[String]] = source.
-      filter((rowMap: Map[String, String]) => rowMap.getOrElse(key = "tconst", default = "") == tvSerieId)
-      .map((rowMap: Map[String, String]) => rowMap.get(key = "tconst"))
+    tvSerieIDFlow
+  }
+
+  def getTvSerieIDFuture(tvSerieID: String, logger: Logger, inputFile: File): Future[Option[String]] = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = inputFile)
+
+    val tvSerieIdFuture: Future[Option[String]] = source
+      .filter(rowMap => rowMap.getOrElse(key = "tconst", default = "") == tvSerieID)
+      .map(row => row.get("tconst"))
       .runWith(Sink.head)
 
+    tvSerieIdFuture.onComplete({
+      case Failure(noSuchElementException: NoSuchElementException) => throw new TvSerieNotFoundException()
+      case Failure(exception: Exception) => throw new RuntimeException()
+      case Success(value) => logger.info(s"$value")
+    })
+
     tvSerieIdFuture
   }