first commit

pom.xml

@@ -66,6 +66,14 @@
             <version>2.1.2</version>
         </dependency>
 
+        <!-- https://mvnrepository.com/artifact/io.spray/spray-json -->
+        <dependency>
+            <groupId>io.spray</groupId>
+            <artifactId>spray-json_2.12</artifactId>
+            <version>1.3.6</version>
+        </dependency>
+
+
     </dependencies>
 
     <build>

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

@@ -1,9 +1,11 @@
 package fr.natan.akkastreamfileprocessingapi.controller;
 
+import fr.natan.akkastreamfileprocessingapi.futurecompleteness.CompletableFutureResult;
 import fr.natan.akkastreamfileprocessingapi.models.Models;
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamFileProcessing;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
@@ -11,6 +13,10 @@ import org.springframework.web.bind.annotation.RestController;
 import scala.collection.IndexedSeq;
 import scala.concurrent.Future;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+
 @SuppressWarnings("SpellCheckingInspection")
 @RestController
 public class MovieController {
@@ -22,29 +28,36 @@ public class MovieController {
     }
 
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
-    private Future<Models.Person> getPersonByID(@PathVariable(name = "personID") String nconst){
-
-       return akkaStreamFilesProcessing.getPersonById(nconst);
+    private ResponseEntity<String> getPersonByID(@PathVariable(name = "personID") String personID) throws ExecutionException, InterruptedException {
+        Future<Models.Person> personFuture = akkaStreamFilesProcessing.getPersonById(personID);
+        CompletableFuture<Models.Person> completableFutureResult =
+                new CompletableFutureResult().buildcompletableFuture(personFuture);
 
+        return new ResponseEntity<>(completableFutureResult.get().toString(), HttpStatus.OK);
     }
 
     @RequestMapping(value = "/persons/name/{primaryName}", method = RequestMethod.GET)
-    private Future<IndexedSeq<Models.Person>> getPersonByName(@PathVariable(name = "primaryName") String primaryName){
-        return akkaStreamFilesProcessing.getPersonByName(primaryName);
+    private ResponseEntity<IndexedSeq<String>> getPersonByName(@PathVariable(name = "primaryName") String primaryName) throws ExecutionException, InterruptedException {
+        Future<IndexedSeq<Models.Person>> personsFuture = akkaStreamFilesProcessing.getPersonByName(primaryName);
+        CompletableFuture completableFutureResult =new
+                CompletableFutureResult().buildCompletableFuture2(personsFuture);
+        IndexedSeq<String> list = (IndexedSeq<String>) completableFutureResult.get();
+        return new ResponseEntity<>(list, HttpStatus.OK);
     }
 
     @RequestMapping(value = "/tvseries/{tvseriePrimaryTitle}", method = RequestMethod.GET)
-    private Future<IndexedSeq<Models.TvSeries>> getTvserieByPrimaryTitle(@PathVariable(name ="tvseriePrimaryTitle" ) String tvseriePrimaryTitle){
+    private Future<IndexedSeq<Models.TvSeries>> getTvserieByPrimaryTitle(@PathVariable(name = "tvseriePrimaryTitle") String tvseriePrimaryTitle) {
         return akkaStreamFilesProcessing.getTvSerieByPrimaryTitle(tvseriePrimaryTitle);
     }
 
     @RequestMapping(value = "/tvseries/title/{tvSerieTitle}", method = RequestMethod.GET)
-    private Future<IndexedSeq<Models.Person>> getPersonsForTvSerie(@PathVariable(name = "tvSerieTitle") String tvSerieTitle){
-      return akkaStreamFilesProcessing.getTeamOfPersonsForTvSerie(tvSerieTitle);
+    @Async
+    private Future<IndexedSeq<Models.Person>> getPersonsForTvSerie(@PathVariable(name = "tvSerieTitle") String tvSerieTitle) {
+        return akkaStreamFilesProcessing.getTeamOfPersonsForTvSerie(tvSerieTitle);
     }
 
     @RequestMapping(value = "/persons", method = RequestMethod.GET)
-    private ResponseEntity<String> getAllPersons(){
+    private ResponseEntity<String> getAllPersons() {
         akkaStreamFilesProcessing.getAllPersons();
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }

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

@@ -1,4 +1,4 @@
-package fr.natan.akkastreamfileprocessingapi.controller;
+package fr.natan.akkastreamfileprocessingapi.exceptionshandler;
 
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException;
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.MovieNotFoundException;

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

@@ -0,0 +1,31 @@
+package fr.natan.akkastreamfileprocessingapi.futurecompleteness;
+
+import scala.collection.IndexedSeq;
+import scala.concurrent.Future;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+
+
+public class CompletableFutureResult<T> {
+
+    public CompletableFuture<T> buildcompletableFuture(Future<T> future){
+        CompletableFuture<T> completableFuture = new CompletableFuture<>();
+        Executors.newCachedThreadPool().submit(()->{
+            Thread.sleep(500);
+            return completableFuture.complete(future.value().get().get());
+        });
+        return completableFuture;
+    }
+
+    public CompletableFuture<IndexedSeq<T>> buildCompletableFuture2(Future<IndexedSeq<T>> futures){
+        CompletableFuture<IndexedSeq<T>> completableFuture = new CompletableFuture<>();
+        Executors.newCachedThreadPool().submit(
+                ()->{
+                    Thread.sleep(500);
+                    return completableFuture.complete(futures.value().get().get().seq());
+                }
+        );
+        return completableFuture;
+    }
+}

src/main/scala/fr/natan/akkastreamfileprocessingapi/models/Models.scala

@@ -2,23 +2,18 @@ package fr.natan.akkastreamfileprocessingapi.models
 
 object Models {
 
-  final case class Person(
-                           nconst: String,
-                           primaryName: String,
-                           birthYear: String,
-                           deathYear: String,
-                           primaryProfession: List[String],
-                           knownForTitles: List[String]
-                         ) {
+  final case class Person(nconst: String, primaryName: String, birthYear: String, deathYear: String,
+                          primaryProfession: List[String], knownForTitles: List[String]) {
     override def toString: String = {
       "Person[person-ID:" + nconst +
         ", primary-name:" + primaryName +
         ", birth-year:" + birthYear +
-        ", dearth year:" + deathYear +
-        ", primary profession:" + primaryProfession +
-      ", known-for-titles"+knownForTitles+
+        ", dearth-year:" + deathYear +
+        ", primary-profession:" + primaryProfession +
+        ", known-for-titles" + knownForTitles +
         "]"
     }
+
   }
 
   final case class TvSeries(

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

@@ -6,7 +6,7 @@ import scala.concurrent.Future
 
 //noinspection SpellCheckingInspection,AccessorLikeMethodIsUnit
 trait AkkaStreamFileProcessing {
-  def getPersonById(nconst: String): Future[Person]
+  def getPersonById(personID: String): Future[Person]
   def getPersonByName(primaryName: String): Future[IndexedSeq[Person]]
 
   def getTvSerieByPrimaryTitle(tvSerieTitle: String):Future[IndexedSeq[TvSeries]]

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

@@ -26,10 +26,10 @@ 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] = {
+  override def getPersonById(personID: String): Future[Person] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    val res = source
-      .via(flow = filterByPersonIdFlow(nconst = nconst))
+    val res: Future[Person] = source
+      .via(flow = filterByPersonIdFlow(nconst = personID))
       .runWith(Sink.head)
 
     res.andThen {
@@ -90,11 +90,11 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     res
   }
 
-  private def getListOfPersonsIDByTvSerieID(tvSerieID: Future[Option[String]]): Future[IndexedSeq[Option[String]]] = {
+  private def getListOfPersonsIDByTvSerieID(tvSerieIdFuture: 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
+        rowMaps.getOrElse(key = "tconst", default = "") == tvSerieIdFuture.value.get.get.get
       })
       .map((rowMap: Map[String, String]) => {
         rowMap.get(key = "nconst")
@@ -104,18 +104,18 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     res
   }
 
-  private def getListOfPersonsForTvSerie(listPersonsIDs: Future[IndexedSeq[Option[String]]]): Future[IndexedSeq[Person]] = {
-    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+  private def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]]):
+  Future[IndexedSeq[Person]] = {
 
-    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)
+    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+    val res: 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)
 
     res
   }
@@ -126,39 +126,39 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     logger.info("STEP 1/3 START")
     val tvSerieIDFuture: Future[Option[String]] = getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
 
-    val finalResult: 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("STEP 1/3 END")
-    })
-      .flatMap({
-       _ =>
-         logger.info("STEP 2/3 START")
-         val listPersonIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieID = tvSerieIDFuture)
-          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("STEP 2/3 END")
-          })
-            .flatMap({
-              future =>
-                logger.info("STEP 3/3 START")
-                val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDs = listPersonIDsFuture)
-                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("STEP 3/3 END")
-                })
-            })
+    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("STEP 1/3 END")
       })
+        .flatMap({
+          _ =>
+            logger.info("STEP 2/3 START")
+            val listPersonIDsFuture: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieIdFuture = tvSerieIDFuture)
+            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("STEP 2/3 END")
+            })
+              .flatMap({
+                future =>
+                  logger.info("STEP 3/3 START")
+                  val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDsFuture = listPersonIDsFuture)
+                  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("STEP 3/3 END")
+                  })
+              })
+        })
 
-    finalResult
+    finalFuture
   }
 
-
   override def getAllPersons(): Unit = {
     val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
     //graph