first commit

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

@@ -1,16 +1,12 @@
 package fr.natan.akkastreamfileprocessingapi.futurecompleteness;
 
-import akka.util.Timeout;
 import scala.Option;
 import scala.collection.IndexedSeq;
-import scala.concurrent.Await;
 import scala.concurrent.Future;
 
-import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 
 public class CompletableFutureBuilder<T> {
@@ -19,9 +15,10 @@ public class CompletableFutureBuilder<T> {
         CompletableFuture<T> completableFuture = new CompletableFuture<>();
         Executors.newSingleThreadExecutor().submit(() -> {
             while(!futureT.isCompleted()){
-                TimeUnit.MILLISECONDS.sleep(100);
+                TimeUnit.MILLISECONDS.sleep(300);
             }
-            completableFuture.complete(futureT.value().get().get().get());
+            T task = futureT.value().get().get().get();
+            completableFuture.complete(task);
             return null;
         });
 
@@ -29,12 +26,15 @@ public class CompletableFutureBuilder<T> {
     }
 
     public static <T> CompletableFuture<IndexedSeq<T>> buildcompletableFuture2(Future<IndexedSeq<T>> futureListT) {
+
         CompletableFuture<IndexedSeq<T>> completableFuture = new CompletableFuture<>();
         Executors.newSingleThreadExecutor().submit(() -> {
             while (!futureListT.isCompleted()) {
                 TimeUnit.MILLISECONDS.sleep(300);
             }
-            completableFuture.complete(futureListT.value().get().get());
+
+            IndexedSeq<T> task = futureListT.value().get().get();
+            completableFuture.complete(task);
             return null;
         });
 

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

@@ -15,6 +15,6 @@ trait AkkaStreamFileProcessingFuture {
 
   def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]
   def getPersonsForTvSerieByTvSerieIDFuture(tvSerieId: String): Future[IndexedSeq[Person]]
-  def getAllTvSeriesFuture(): Future[Done]
-  def getAllPersonsFuture(): Future[Done]
+  def getAllTvSeriesFuture: Future[Done]
+  def getAllPersonsFuture: Future[Done]
 }

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

@@ -7,11 +7,9 @@ import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, t
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSerie}
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
   actorSystem, buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildSource,
-  buildTvSerieFlow, filterPersonByIdFlow, filterPersonByNameFlow, filterTvSerieByIdFlow, filterTvSerieByPrimaryTitleFlow
-}
+  buildTvSerieFlow, filterPersonByIdFlow, filterPersonByNameFlow, filterTvSerieByIdFlow, filterTvSerieByPrimaryTitleFlow}
 import fr.natan.akkastreamfileprocessingapi.service.UtilitiesClass.{
-  getListOfPersonsForTvSerie, getListOfPersonsIDByTvSerieID, getTvSerieIDFuture, getTvSerieIdByPrimaryTitle
-}
+  getListOfPersonsForTvSerie, getListOfPersonsIDByTvSerieID, getTvSerieIDFuture, getTvSerieIdByPrimaryTitle}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -32,11 +30,11 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     val start: Long = System.currentTimeMillis()
     val personFuture: Future[Option[Person]] = source
       .via(flow = filterPersonByIdFlow(personID = personID))
-      .runWith(Sink.headOption)
+      .runWith(Sink.headOption[Person])
 
     personFuture.andThen({
       case Failure(exception) => logger.info(s"$exception")
-      case Success(value:Option[Person]) =>
+      case Success(value: Option[Person]) =>
         logger.info(s"$value")
         logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
     })
@@ -44,13 +42,13 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     personFuture
   }
 
-  override def getPersonByNameFuture(primaryName: String):  Future[Option[Person]]= {
+  override def getPersonByNameFuture(primaryName: String): Future[Option[Person]] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
 
     val start: Long = System.currentTimeMillis()
-    val personFuture:  Future[Option[Person]] = source
+    val personFuture: Future[Option[Person]] = source
       .via(flow = filterPersonByNameFlow(primaryName = primaryName))
-      .runWith(Sink.headOption)
+      .runWith(Sink.headOption[Person])
 
     personFuture
   }
@@ -59,9 +57,9 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
 
     val start: Long = System.currentTimeMillis()
-    val tvSerieFuture: Future[Option[TvSerie]]= source
+    val tvSerieFuture: Future[Option[TvSerie]] = source
       .via(flow = filterTvSerieByIdFlow(tvSerieID = tvSerieID))
-      .runWith(Sink.headOption)
+      .runWith(Sink.headOption[TvSerie])
 
     tvSerieFuture.onComplete({
       case Failure(exception) => logger.info(s"$exception")
@@ -71,6 +69,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     })
     tvSerieFuture
   }
+
   override def getTvSeriesByPrimaryTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[TvSerie]] = {
 
     val start: Long = System.currentTimeMillis()
@@ -102,11 +101,12 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
         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")
+
       })
         .flatMap({
           _ =>
             logger.info("STEP 2/3 START")
-            val  start2: Long = System.currentTimeMillis()
+            val start2: Long = System.currentTimeMillis()
             val listPersonIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
               tvSerieIdFuture = tvSerieIDFuture)
             listPersonIDsFuture.andThen({
@@ -140,15 +140,16 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     val tvSerieIdFuture: Future[Option[String]] = getTvSerieIDFuture(tvSerieId = tvSerieId)
     tvSerieIdFuture.andThen({
       case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
-      case Success(value:Option[String]) =>
+      case Success(value: Option[String]) =>
         logger.info(s"$value")
         logger.info(s"STEP 1/3 END, elapsed time:${(System.currentTimeMillis() - start1) / 1000} sec")
 
     })
       .flatMap({
-        _ => logger.info("STEP 2/3")
+        _ =>
+          logger.info("STEP 2/3")
           val start2: Long = System.currentTimeMillis()
-          val  listOfPersonsIDsFuture : Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
+          val listOfPersonsIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(
             tvSerieIdFuture = tvSerieIdFuture)
           listOfPersonsIDsFuture.andThen({
             case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
@@ -157,7 +158,8 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
               logger.info(s"STEP 2/3 END, elapsed time:${(System.currentTimeMillis() - start2) / 1000} sec")
           })
             .flatMap({
-              _ => logger.info("STEP 3/3")
+              _ =>
+                logger.info("STEP 3/3")
                 val start3: Long = System.currentTimeMillis()
                 val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(
                   listPersonsIDsFuture = listOfPersonsIDsFuture)
@@ -171,7 +173,8 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
       })
 
   }
-  override def getAllPersonsFuture(): Future[Done] = {
+
+  override def getAllPersonsFuture: Future[Done] = {
     val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
     //graph
     val startTime: Long = System.currentTimeMillis()
@@ -190,7 +193,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     result
   }
 
-  override def getAllTvSeriesFuture(): Future[Done] = {
+  override def getAllTvSeriesFuture: Future[Done] = {
     val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
     val sink: Sink[TvSerie, Future[Done]] = buildAllTvSeriesSink(logger = logger)