first commit

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

@@ -13,6 +13,7 @@ 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;
 
@@ -22,6 +23,8 @@ import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
+import java.util.concurrent.TimeoutException;
+
 
 @SuppressWarnings("SpellCheckingInspection")
 @RestController
@@ -29,7 +32,7 @@ public class TvSeriesController {
 
     private final AkkaStreamFileProcessingFuture akkaStreamFilesProcessing;
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
-    private static final int BORN_INF = 500, BORN_SUP = 5000;
+    private static final int BORN_INF = 100, BORN_SUP = 2000;
 
     public TvSeriesController(AkkaStreamFileProcessingFuture akkaStreamFilesProcessing) {
         this.akkaStreamFilesProcessing = akkaStreamFilesProcessing;
@@ -37,7 +40,7 @@ public class TvSeriesController {
 
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
     private String getPersonByID(@PathVariable(name = "personID") String personID) throws ExecutionException, InterruptedException {
-        Future<Models.Person> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
+        Future<Option<Models.Person>> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
         CompletableFuture<Models.Person> completableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(futurePerson);
         while (!completableFuture.isDone()) {
@@ -45,6 +48,7 @@ public class TvSeriesController {
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
         }
         Models.Person person = completableFuture.get();
+
         JsValue personJs = Json.toJson(person, person.personFormat());
         return Json.prettyPrint(personJs);
     }
@@ -53,7 +57,7 @@ public class TvSeriesController {
     private String getPersonByName(@PathVariable(name = "primaryName") String primaryName)
             throws ExecutionException, InterruptedException {
 
-        Future<Models.Person> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
+        Future<Option<Models.Person>> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
         CompletableFuture<Models.Person> personCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(personFuture);
         while (!personCompletableFuture.isDone()) {
@@ -68,7 +72,7 @@ public class TvSeriesController {
     @RequestMapping(value = "/tvseries/id/{tvSerieId}", method = RequestMethod.GET)
     private String getTvSerieByID(@PathVariable(name = "tvSerieId") String tvSerieID)
             throws ExecutionException, InterruptedException {
-        Future<Models.TvSerie> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
+        Future<Option<Models.TvSerie>> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
         CompletableFuture<Models.TvSerie> tvSerieCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(tvSerieFuture);
         while (!tvSerieCompletableFuture.isDone()) {
@@ -137,7 +141,7 @@ public class TvSeriesController {
                 .buildcompletableFuture2(futureIndexedSeqPersons);
 
         while (!seqCompletableFuture.isDone()){
-            logger.info("IS PROCESSING");
+            logger.info("IS PROCESSING...");
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
         }
         IndexedSeq<Models.Person> personIndexedSeq = seqCompletableFuture.get();

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

@@ -1,35 +1,38 @@
 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.util.Random;
+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> {
 
-    public static<T> CompletableFuture<T> buildcompletableFuture1(Future<T> futureT) {
+    public static<T> CompletableFuture<T> buildcompletableFuture1(Future<Option<T>> futureT) {
         CompletableFuture<T> completableFuture = new CompletableFuture<>();
-       Executors.newSingleThreadExecutor().submit(() -> {
-           while(!futureT.isCompleted()){
-               Thread.sleep(new Random().nextInt(500-100)+100);
-           }
-            completableFuture.complete(futureT.value().get().get());
+        Executors.newSingleThreadExecutor().submit(() -> {
+            while(!futureT.isCompleted()){
+                TimeUnit.MILLISECONDS.sleep(100);
+            }
+            completableFuture.complete(futureT.value().get().get().get());
             return null;
         });
 
         return completableFuture;
     }
 
-    public static<T> CompletableFuture<IndexedSeq<T>> buildcompletableFuture2(Future<IndexedSeq<T>> futureListT) {
+    public static <T> CompletableFuture<IndexedSeq<T>> buildcompletableFuture2(Future<IndexedSeq<T>> futureListT) {
         CompletableFuture<IndexedSeq<T>> completableFuture = new CompletableFuture<>();
         Executors.newSingleThreadExecutor().submit(() -> {
-            while(!futureListT.isCompleted()){
-                Thread.sleep(100);
+            while (!futureListT.isCompleted()) {
+                TimeUnit.MILLISECONDS.sleep(300);
             }
             completableFuture.complete(futureListT.value().get().get());
             return null;

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

@@ -7,10 +7,10 @@ import scala.concurrent.Future
 
 //noinspection SpellCheckingInspection,AccessorLikeMethodIsUnit
 trait AkkaStreamFileProcessingFuture {
-  def getPersonByIdFuture(personID: String): Future[Person]
-  def getPersonByNameFuture(primaryName: String): Future[Person]
+  def getPersonByIdFuture(personID: String): Future[Option[Person]]
+  def getPersonByNameFuture(primaryName: String):  Future[Option[Person]]
 
-  def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie]
+  def getTvSerieByIdFuture(tvSerieID: String): Future[Option[TvSerie]]
   def getTvSeriesByPrimaryTitleFuture(tvSerieTitle: String):Future[IndexedSeq[TvSerie]]
 
   def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]

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

@@ -6,23 +6,11 @@ import com.typesafe.scalalogging.slf4j.Logger
 import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics}
 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
+  actorSystem, buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildSource,
+  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
@@ -38,17 +26,17 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
 
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getPersonByIdFuture(personID: String): Future[Person] = {
+  override def getPersonByIdFuture(personID: String): Future[Option[Person]] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
 
     val start: Long = System.currentTimeMillis()
-    val personFuture: Future[Person] = source
+    val personFuture: Future[Option[Person]] = source
       .via(flow = filterPersonByIdFlow(personID = personID))
-      .runWith(Sink.head)
+      .runWith(Sink.headOption)
 
     personFuture.andThen({
       case Failure(exception) => logger.info(s"$exception")
-      case Success(value:Person) =>
+      case Success(value:Option[Person]) =>
         logger.info(s"$value")
         logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
     })
@@ -56,34 +44,28 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
     personFuture
   }
 
-  override def getPersonByNameFuture(primaryName: String): Future[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[Person] = source
+    val personFuture:  Future[Option[Person]] = source
       .via(flow = filterPersonByNameFlow(primaryName = primaryName))
-      .runWith(Sink.head)
-
-    personFuture.onComplete({
-      case Failure(exception) => logger.error(s"$exception")
-      case Success(value: Person) => logger.info(s"$value")
-        logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
-    })
+      .runWith(Sink.headOption)
 
     personFuture
   }
 
-  override def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie] = {
+  override def getTvSerieByIdFuture(tvSerieID: String): Future[Option[TvSerie]] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
 
     val start: Long = System.currentTimeMillis()
-    val tvSerieFuture: Future[TvSerie] = source
+    val tvSerieFuture: Future[Option[TvSerie]]= source
       .via(flow = filterTvSerieByIdFlow(tvSerieID = tvSerieID))
-      .runWith(Sink.head)
+      .runWith(Sink.headOption)
 
     tvSerieFuture.onComplete({
       case Failure(exception) => logger.info(s"$exception")
-      case Success(value: TvSerie) =>
+      case Success(value: Option[TvSerie]) =>
         logger.info(s"$value")
         logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
     })