first commit

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

@@ -84,7 +84,8 @@ public class TvSeriesController {
     private List<String> getTvserieByPrimaryTitle(
             @PathVariable(name = "tvseriePrimaryTitle") String tvseriePrimaryTitle) throws InterruptedException, ExecutionException {
 
-        Future<IndexedSeq<Models.TvSerie>> listFuture = akkaStreamFilesProcessing.getTvSeriesByPrimaryTitleFuture(tvseriePrimaryTitle);
+        Future<IndexedSeq<Models.TvSerie>> listFuture = akkaStreamFilesProcessing
+                .getTvSeriesByPrimaryTitleFuture(tvseriePrimaryTitle);
         CompletableFuture<IndexedSeq<Models.TvSerie>> completableFuture =
                 CompletableFutureBuilder.buildcompletableFuture2(listFuture);
         while (!completableFuture.isDone()) {
@@ -106,8 +107,10 @@ public class TvSeriesController {
     private List<String> getPersonsForTvSerieByTvSerieTitle(@PathVariable(name = "tvSerieTitle") String tvSerieTitle)
             throws InterruptedException, ExecutionException {
 
-        Future<IndexedSeq<Models.Person>> futurePersonSeq = akkaStreamFilesProcessing.getPersonsForTvSerieByTvSerieTitleFuture(tvSerieTitle);
-        CompletableFuture<IndexedSeq<Models.Person>> completableFuture = CompletableFutureBuilder.buildcompletableFuture2(futurePersonSeq);
+        Future<IndexedSeq<Models.Person>> futurePersonSeq = akkaStreamFilesProcessing
+                .getPersonsForTvSerieByTvSerieTitleFuture(tvSerieTitle);
+        CompletableFuture<IndexedSeq<Models.Person>> completableFuture = CompletableFutureBuilder
+                .buildcompletableFuture2(futurePersonSeq);
         while (!completableFuture.isDone()){
             logger.info("IS PROCESSING...");
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
@@ -128,8 +131,10 @@ public class TvSeriesController {
     private List<String> getPersonsForTvSerieByTvSerieID(@PathVariable(name = "tvSerieID") String tvSerieID)
             throws InterruptedException, ExecutionException {
 
-        Future<IndexedSeq<Models.Person>> futureIndexedSeqPersons = akkaStreamFilesProcessing.getPersonsForTvSerieByTvSerieIdFuture(tvSerieID);
-        CompletableFuture<IndexedSeq<Models.Person>> seqCompletableFuture = CompletableFutureBuilder.buildcompletableFuture2(futureIndexedSeqPersons);
+        Future<IndexedSeq<Models.Person>> futureIndexedSeqPersons = akkaStreamFilesProcessing
+                .getPersonsForTvSerieByTvSerieIDFuture(tvSerieID);
+        CompletableFuture<IndexedSeq<Models.Person>> seqCompletableFuture = CompletableFutureBuilder
+                .buildcompletableFuture2(futureIndexedSeqPersons);
 
         while (!seqCompletableFuture.isDone()){
             logger.info("IS PROCESSING");

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

@@ -17,7 +17,7 @@ import scala.concurrent.Future
 //noinspection SpellCheckingInspection
 object AkkaStreamComponents {
 
-  implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")
+  implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
 
   //flows building
   def buildPersonFlow(): Flow[Map[String, String], Person, NotUsed] = {

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

@@ -14,7 +14,7 @@ trait AkkaStreamFileProcessingFuture {
   def getTvSeriesByPrimaryTitleFuture(tvSerieTitle: String):Future[IndexedSeq[TvSerie]]
 
   def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]
-  def getPersonsForTvSerieByTvSerieIdFuture(tvSerieId: String): Future[IndexedSeq[Person]]
+  def getPersonsForTvSerieByTvSerieIDFuture(tvSerieId: String): Future[IndexedSeq[Person]]
   def getAllTvSeriesFuture(): Future[Done]
   def getAllPersonsFuture(): Future[Done]
 }

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

@@ -1,13 +1,12 @@
 package fr.natan.akkastreamfileprocessingapi.service
 
 import akka.Done
-import akka.actor.ActorSystem
 import akka.stream.scaladsl.{Sink, Source}
 import com.typesafe.scalalogging.slf4j.Logger
-import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
+import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics}
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSerie}
-import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildPersonModel
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
+  actorSystem,
   buildAllPersonsSink,
   buildAllTvSeriesSink,
   buildAndValidateSource,
@@ -19,6 +18,12 @@ import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
   filterTvSerieByIdFlow,
   filterTvSerieByPrimaryTitleFlow
 }
+import fr.natan.akkastreamfileprocessingapi.service.UtilitiesClass.{
+  getListOfPersonsForTvSerie,
+  getListOfPersonsIDByTvSerieID,
+  getTvSerieIDFuture,
+  getTvSerieIdByPrimaryTitle
+}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -31,7 +36,6 @@ import scala.util.{Failure, Success}
 @Component
 class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
 
-  implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
   override def getPersonByIdFuture(personID: String): Future[Person] = {
@@ -103,48 +107,6 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     tvSriesFuture
   }
 
-
-  private def getTvSerieIdByPrimaryTitle(primaryTitle: String): Future[Option[String]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
-    val tvSerieIdFuture: Future[Option[String]] = source
-      .filter((rowMap: Map[String, String]) => {
-        rowMap.getOrElse(key = "primaryTitle", default = "") == primaryTitle
-      })
-      .map((rowMap: Map[String, String]) => rowMap.get("tconst"))
-      .runWith(Sink.head)
-
-    tvSerieIdFuture
-  }
-
-  private def getListOfPersonsIDByTvSerieID(tvSerieIdFuture: Future[Option[String]]): Future[IndexedSeq[Option[String]]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
-    val personsIDsFuture: Future[IndexedSeq[Option[String]]] = source
-      .filter((rowMaps: Map[String, String]) => {
-        rowMaps.getOrElse(key = "tconst", default = "") == tvSerieIdFuture.value.get.get.get
-      })
-      .map((rowMap: Map[String, String]) => {
-        rowMap.get(key = "nconst")
-      })
-      .runWith(Sink.collection)
-
-    personsIDsFuture
-  }
-
-  private def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]]): Future[IndexedSeq[Person]] = {
-
-    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    val personsFuture: Future[IndexedSeq[Person]] = source
-      .filter((rowMaps: Map[String, String]) => {
-        listPersonsIDsFuture.value.get.get.contains(rowMaps.get(key = "nconst"))
-      })
-      .map((rowMap: Map[String, String]) => {
-        buildPersonModel(rowMap)
-      })
-      .runWith(Sink.collection)
-
-    personsFuture
-  }
-
   override def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]] = {
 
     //futures chaining
@@ -163,7 +125,8 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
           _ =>
             logger.info("STEP 2/3 START")
             val  start2: Long = System.currentTimeMillis()
-            val listPersonIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieIdFuture = tvSerieIDFuture)
+            val listPersonIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
+              tvSerieIdFuture = tvSerieIDFuture)
             listPersonIDsFuture.andThen({
               case Failure(exception) => logger.error(s"$exception")
               case Success(value) =>
@@ -174,7 +137,8 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
                 _ =>
                   logger.info("STEP 3/3 START")
                   val start3: Long = System.currentTimeMillis()
-                  val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDsFuture = listPersonIDsFuture)
+                  val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(
+                    listPersonsIDsFuture = listPersonIDsFuture)
                   personsTeamFuture.andThen({
                     case Failure(exception) => logger.error(s"$exception")
                     case Success(value: IndexedSeq[Person]) =>
@@ -187,18 +151,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     finalFuture
   }
 
-  private def getTvSerieIDFuture(tvSerieId: String): Future[Option[String]] ={
-    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
-
-    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"))
-      .runWith(Sink.head)
-
-    tvSerieIdFuture
-  }
-
-  override def getPersonsForTvSerieByTvSerieIdFuture(tvSerieId: String): Future[IndexedSeq[Person]] = {
+  override def getPersonsForTvSerieByTvSerieIDFuture(tvSerieId: String): Future[IndexedSeq[Person]] = {
 
     val start1: Long = System.currentTimeMillis()
     logger.info("STEP 1/3")
@@ -213,17 +166,19 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
       .flatMap({
         _ => logger.info("STEP 2/3")
           val start2: Long = System.currentTimeMillis()
-          val  listOfPersonsIDsFuture : Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieIdFuture = tvSerieIdFuture)
+          val  listOfPersonsIDsFuture : Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
+            tvSerieIdFuture = tvSerieIdFuture)
           listOfPersonsIDsFuture.andThen({
             case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
-            case Success(value) =>
+            case Success(value: IndexedSeq[Option[String]]) =>
               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")
                 val start3: Long = System.currentTimeMillis()
-                val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDsFuture = listOfPersonsIDsFuture)
+                val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(
+                  listPersonsIDsFuture = listOfPersonsIDsFuture)
                 personsTeamFuture.andThen({
                   case Failure(exception) => logger.error(s"$exception")
                   case Success(value: IndexedSeq[Person]) =>

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

@@ -0,0 +1,65 @@
+package fr.natan.akkastreamfileprocessingapi.service
+
+import akka.stream.scaladsl.{Sink, Source}
+import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
+import fr.natan.akkastreamfileprocessingapi.models.Models.Person
+import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildPersonModel
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{actorSystem, buildSource}
+
+import scala.concurrent.Future
+
+object UtilitiesClass {
+
+  def getTvSerieIdByPrimaryTitle(primaryTitle: String): Future[Option[String]] = {
+    val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
+    val tvSerieIdFuture: Future[Option[String]] = source
+      .filter((rowMap: Map[String, String]) => {
+        rowMap.getOrElse(key = "primaryTitle", default = "") == primaryTitle
+      })
+      .map((rowMap: Map[String, String]) => rowMap.get("tconst"))
+      .runWith(Sink.head)
+
+    tvSerieIdFuture
+  }
+
+  def getListOfPersonsIDByTvSerieID(tvSerieIdFuture: Future[Option[String]]): Future[IndexedSeq[Option[String]]] = {
+    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
+    val personsIDsFuture: Future[IndexedSeq[Option[String]]] = source
+      .filter((rowMaps: Map[String, String]) => {
+        rowMaps.getOrElse(key = "tconst", default = "") == tvSerieIdFuture.value.get.get.get
+      })
+      .map((rowMap: Map[String, String]) => {
+        rowMap.get(key = "nconst")
+      })
+      .runWith(Sink.collection)
+
+    personsIDsFuture
+  }
+
+  def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]]): Future[IndexedSeq[Person]] = {
+
+    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+    val personsFuture: Future[IndexedSeq[Person]] = source
+      .filter((rowMaps: Map[String, String]) => {
+        listPersonsIDsFuture.value.get.get.contains(rowMaps.get(key = "nconst"))
+      })
+      .map((rowMap: Map[String, String]) => {
+        buildPersonModel(rowMap)
+      })
+      .runWith(Sink.collection)
+
+    personsFuture
+  }
+
+  def getTvSerieIDFuture(tvSerieId: String): Future[Option[String]] = {
+    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
+
+    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"))
+      .runWith(Sink.head)
+
+    tvSerieIdFuture
+  }
+
+}