first commit

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

@@ -1,8 +1,8 @@
 package fr.natan.akkastreamfileprocessingapi.controller;
 
-import fr.natan.akkastreamfileprocessingapi.futurecompleteness.CompletableFutureResult;
+import fr.natan.akkastreamfileprocessingapi.futurecompleteness.CompletableFutureBuilder;
 import fr.natan.akkastreamfileprocessingapi.models.Models;
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamFileProcessing;
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamFileProcessingFuture;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpStatus;
@@ -27,18 +27,19 @@ import java.util.concurrent.ExecutionException;
 @RestController
 public class TvSeriesController {
 
-    private final AkkaStreamFileProcessing akkaStreamFilesProcessing;
+    private final AkkaStreamFileProcessingFuture akkaStreamFilesProcessing;
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
     private static final int BORN_INF = 500, BORN_SUP = 5000;
 
-    public TvSeriesController(AkkaStreamFileProcessing akkaStreamFilesProcessing) {
+    public TvSeriesController(AkkaStreamFileProcessingFuture akkaStreamFilesProcessing) {
         this.akkaStreamFilesProcessing = akkaStreamFilesProcessing;
     }
 
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
     private String getPersonByID(@PathVariable(name = "personID") String personID) throws ExecutionException, InterruptedException {
-        Future<Models.Person> futurePerson = akkaStreamFilesProcessing.getPersonById(personID);
-        CompletableFuture<Models.Person> completableFuture = CompletableFutureResult.buildcompletableFuture1(futurePerson);
+        Future<Models.Person> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
+        CompletableFuture<Models.Person> completableFuture = CompletableFutureBuilder
+                .buildcompletableFuture1(futurePerson);
         while (!completableFuture.isDone()) {
             logger.info("IS PROCESSING...");
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
@@ -52,8 +53,9 @@ public class TvSeriesController {
     private String getPersonByName(@PathVariable(name = "primaryName") String primaryName)
             throws ExecutionException, InterruptedException {
 
-        Future<Models.Person> personFuture = akkaStreamFilesProcessing.getPersonByName(primaryName);
-        CompletableFuture<Models.Person> personCompletableFuture = CompletableFutureResult.buildcompletableFuture1(personFuture);
+        Future<Models.Person> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
+        CompletableFuture<Models.Person> personCompletableFuture = CompletableFutureBuilder
+                .buildcompletableFuture1(personFuture);
         while (!personCompletableFuture.isDone()) {
             logger.info("IS PROCESSING...");
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
@@ -65,8 +67,9 @@ 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.getTvSerieById(tvSerieID);
-        CompletableFuture<Models.TvSerie> tvSerieCompletableFuture = CompletableFutureResult.buildcompletableFuture1(tvSerieFuture);
+        Future<Models.TvSerie> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
+        CompletableFuture<Models.TvSerie> tvSerieCompletableFuture = CompletableFutureBuilder
+                .buildcompletableFuture1(tvSerieFuture);
         while (!tvSerieCompletableFuture.isDone()) {
             logger.info("IS PROCESSING...");
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
@@ -79,9 +82,9 @@ public class TvSeriesController {
     private List<String> getTvserieByPrimaryTitle(
             @PathVariable(name = "tvseriePrimaryTitle") String tvseriePrimaryTitle) throws InterruptedException, ExecutionException {
 
-        Future<IndexedSeq<Models.TvSerie>> listFuture = akkaStreamFilesProcessing.getTvSerieByPrimaryTitle(tvseriePrimaryTitle);
+        Future<IndexedSeq<Models.TvSerie>> listFuture = akkaStreamFilesProcessing.getTvSeriesByPrimaryTitleFuture(tvseriePrimaryTitle);
         CompletableFuture<IndexedSeq<Models.TvSerie>> completableFuture =
-                CompletableFutureResult.buildcompletableFuture2(listFuture);
+                CompletableFutureBuilder.buildcompletableFuture2(listFuture);
         while (!completableFuture.isDone()) {
             logger.info("IS PROCESSING...");
             Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
@@ -98,11 +101,11 @@ public class TvSeriesController {
     }
 
     @RequestMapping(value = "/persons/tvseries/title/{tvSerieTitle}", method = RequestMethod.GET)
-    private List<String> getPersonsForTvSerie(@PathVariable(name = "tvSerieTitle") String tvSerieTitle)
+    private List<String> getPersonsForTvSerieByTvSerieTitle(@PathVariable(name = "tvSerieTitle") String tvSerieTitle)
             throws InterruptedException, ExecutionException {
 
-        Future<IndexedSeq<Models.Person>> futurePersonSeq = akkaStreamFilesProcessing.getTeamOfPersonsForTvSerie(tvSerieTitle);
-        CompletableFuture<IndexedSeq<Models.Person>> completableFuture = CompletableFutureResult.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);
@@ -119,15 +122,36 @@ public class TvSeriesController {
         return persons;
     }
 
+    @RequestMapping(value = "/persons/tvseries/id/{tvSerieID}", method = RequestMethod.GET)
+    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);
+
+        while (!seqCompletableFuture.isDone()){
+            logger.info("IS PROCESSING");
+            Thread.sleep(new Random().nextInt(BORN_SUP-BORN_INF)+BORN_INF);
+        }
+        IndexedSeq<Models.Person> personIndexedSeq = seqCompletableFuture.get();
+        List<String> personsList = new ArrayList<>();
+        personIndexedSeq.foreach(person ->{
+            personsList.add(Json.toJson(person, person.personFormat()).toString());
+            return null;
+        });
+
+        return personsList;
+    }
+
     @RequestMapping(value = "/persons", method = RequestMethod.GET)
     private ResponseEntity<String> getAllPersons() {
-        akkaStreamFilesProcessing.getAllPersons();
+        akkaStreamFilesProcessing.getAllPersonsFuture();
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 
     @RequestMapping(value = "/tvseries", method = RequestMethod.GET)
     private ResponseEntity<String> getAllvSeries() {
-        akkaStreamFilesProcessing.getAllTvSeries();
+        akkaStreamFilesProcessing.getAllTvSeriesFuture();
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 }

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

@@ -9,7 +9,7 @@ import java.util.concurrent.Executors;
 
 
 
-public class CompletableFutureResult<T> {
+public class CompletableFutureBuilder<T> {
 
     public static<T> CompletableFuture<T> buildcompletableFuture1(Future<T> futureT) {
         CompletableFuture<T> completableFuture = new CompletableFuture<>();

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

@@ -1,18 +0,0 @@
-package fr.natan.akkastreamfileprocessingapi.service
-
-import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSerie}
-
-import scala.concurrent.Future
-
-//noinspection SpellCheckingInspection,AccessorLikeMethodIsUnit
-trait AkkaStreamFileProcessing {
-  def getPersonById(personID: String): Future[Person]
-  def getPersonByName(primaryName: String): Future[Person]
-
-  def getTvSerieById(tvSerieID: String): Future[TvSerie]
-  def getTvSerieByPrimaryTitle(tvSerieTitle: String):Future[IndexedSeq[TvSerie]]
-
-  def getTeamOfPersonsForTvSerie(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]
-  def getAllTvSeries(): Unit
-  def getAllPersons(): Unit
-}

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

@@ -0,0 +1,20 @@
+package fr.natan.akkastreamfileprocessingapi.service
+
+import akka.Done
+import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSerie}
+
+import scala.concurrent.Future
+
+//noinspection SpellCheckingInspection,AccessorLikeMethodIsUnit
+trait AkkaStreamFileProcessingFuture {
+  def getPersonByIdFuture(personID: String): Future[Person]
+  def getPersonByNameFuture(primaryName: String): Future[Person]
+
+  def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie]
+  def getTvSeriesByPrimaryTitleFuture(tvSerieTitle: String):Future[IndexedSeq[TvSerie]]
+
+  def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: 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

@@ -4,10 +4,23 @@ 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, titlePrincipalsBasics
+}
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSerie}
 import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildPersonModel
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildSource, buildTvSerieFlow, filterPersonByIdFlow, filterPersonByNameFlow, filterTvSerieByIdFlow, filterTvSerieByPrimaryTitleFlow}
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
+  buildAllPersonsSink,
+  buildAllTvSeriesSink,
+  buildAndValidateSource,
+  buildPersonFlow,
+  buildSource,
+  buildTvSerieFlow,
+  filterPersonByIdFlow,
+  filterPersonByNameFlow,
+  filterTvSerieByIdFlow,
+  filterTvSerieByPrimaryTitleFlow
+}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -18,87 +31,87 @@ import scala.util.{Failure, Success}
 
 //noinspection SpellCheckingInspection
 @Component
-class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
+class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessingFuture {
 
   implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getPersonById(personID: String): Future[Person] = {
+  override def getPersonByIdFuture(personID: String): Future[Person] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    val res: Future[Person] = source
+    val personFuture: Future[Person] = source
       .via(flow = filterPersonByIdFlow(personID = personID))
       .runWith(Sink.head)
 
-    res.andThen({
+    personFuture.andThen({
       case Failure(exception) => logger.info(s"$exception")
       case Success(value:Person) => logger.info(s"$value")
     })
 
-    res
+    personFuture
   }
 
-  override def getPersonByName(primaryName: String): Future[Person] = {
+  override def getPersonByNameFuture(primaryName: String): Future[Person] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    val persons: Future[Person] = source
+    val personFuture: Future[Person] = source
       .via(flow = filterPersonByNameFlow(primaryName = primaryName))
       .runWith(Sink.head)
 
-    persons.onComplete({
+    personFuture.onComplete({
       case Failure(exception) => logger.error(s"$exception")
       case Success(value: Person) => logger.info(s"$value")
         logger.info("SUCCESS")
     })
 
-    persons
+    personFuture
   }
 
-  override def getTvSerieById(tvSerieID: String): Future[TvSerie] = {
+  override def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
-    val res: Future[TvSerie] = source
+    val tvSerieFuture: Future[TvSerie] = source
       .via(flow = filterTvSerieByIdFlow(tvSerieID = tvSerieID))
       .runWith(Sink.head)
 
-    res.onComplete({
+    tvSerieFuture.onComplete({
       case Failure(exception) => logger.info(s"$exception")
       case Success(value: TvSerie) => logger.info(s"$value")
     })
-    res
+    tvSerieFuture
   }
-  override def getTvSerieByPrimaryTitle(tvSeriePrimaryTitle: String): Future[IndexedSeq[TvSerie]] = {
+  override def getTvSeriesByPrimaryTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[TvSerie]] = {
 
     val tvSeriesSource: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
 
-    val tvSries: Future[IndexedSeq[TvSerie]] = tvSeriesSource
+    val tvSriesFuture: Future[IndexedSeq[TvSerie]] = tvSeriesSource
       .via(flow = filterTvSerieByPrimaryTitleFlow(tvSeriePrimaryTitle = tvSeriePrimaryTitle))
       .runWith(Sink.collection)
 
-    tvSries.onComplete({
+    tvSriesFuture.onComplete({
       case Failure(exception) => logger.info(s"$exception")
       case Success(value: IndexedSeq[TvSerie]) =>
         value.foreach((tvSrie: TvSerie) => logger.info(s"$tvSrie"))
         logger.info("SUCCESS")
     })
 
-    tvSries
+    tvSriesFuture
   }
 
 
   private def getTvSerieIdByPrimaryTitle(primaryTitle: String): Future[Option[String]] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
 
-    val res: Future[Option[String]] = source
+    val tvSerieIdFuture: Future[Option[String]] = source
       .filter((rowMap: Map[String, String]) => {
-        rowMap.getOrElse(key = "primaryTitle", "") == primaryTitle
+        rowMap.getOrElse(key = "primaryTitle", default = "") == primaryTitle
       })
       .map((rowMap: Map[String, String]) => rowMap.get("tconst"))
       .runWith(Sink.head)
 
-    res
+    tvSerieIdFuture
   }
 
   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
+    val personsIDsFuture: Future[IndexedSeq[Option[String]]] = source
       .filter((rowMaps: Map[String, String]) => {
         rowMaps.getOrElse(key = "tconst", default = "") == tvSerieIdFuture.value.get.get.get
       })
@@ -107,14 +120,13 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
       })
       .runWith(Sink.collection)
 
-    res
+    personsIDsFuture
   }
 
-  private def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]]):
-  Future[IndexedSeq[Person]] = {
+  private def getListOfPersonsForTvSerie(listPersonsIDsFuture: Future[IndexedSeq[Option[String]]]): Future[IndexedSeq[Person]] = {
 
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    val res: Future[IndexedSeq[Person]] = source
+    val personsFuture: Future[IndexedSeq[Person]] = source
       .filter((rowMaps: Map[String, String]) => {
         listPersonsIDsFuture.value.get.get.contains(rowMaps.get(key = "nconst"))
       })
@@ -123,10 +135,10 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
       })
       .runWith(Sink.collection)
 
-    res
+    personsFuture
   }
 
-  override def getTeamOfPersonsForTvSerie(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]] = {
+  override def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]] = {
 
     //futures chaining
     logger.info("STEP 1/3 START")
@@ -150,7 +162,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
                 logger.info("STEP 2/3 END")
             })
               .flatMap({
-                future =>
+                _ =>
                   logger.info("STEP 3/3 START")
                   val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDsFuture = listPersonIDsFuture)
                   personsTeamFuture.andThen({
@@ -165,7 +177,51 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     finalFuture
   }
 
-  override def getAllPersons(): Unit = {
+  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]] = {
+
+    val tvSerieIdFuture: Future[Option[String]] = getTvSerieIDFuture(tvSerieId = tvSerieId)
+    logger.info("STEP 1/3")
+    tvSerieIdFuture.andThen({
+      case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
+      case Success(value:Option[String]) =>
+        logger.info(s"$value")
+        logger.info("STEP 1/3")
+
+    })
+      .flatMap({
+        _ => logger.info("STEP 2/3")
+          val  listOfPersonsIDsFuture : Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieIdFuture = tvSerieIdFuture)
+          listOfPersonsIDsFuture.andThen({
+            case Failure(exception) => logger.error(s"${exception.printStackTrace()}")
+            case Success(value) =>
+              value.foreach((personID: Option[String]) => logger.info(s"Person ID:$personID"))
+              logger.info("STEP 2/3 END")
+          })
+            .flatMap({
+              _ => logger.info("STEP 3/3")
+                val personsTeamFuture: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDsFuture = listOfPersonsIDsFuture)
+                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")
+                })
+            })
+      })
+
+  }
+  override def getAllPersonsFuture(): Future[Done] = {
     val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
     //graph
     val startTime: Long = System.currentTimeMillis()
@@ -180,15 +236,17 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
         val time: Long = (System.currentTimeMillis() - startTime) / 100
         logger.info(s"elapsed time: $time")
     }
+
+    result
   }
 
-  override def getAllTvSeries(): Unit = {
+  override def getAllTvSeriesFuture(): Future[Done] = {
     val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
     val sink: Sink[TvSerie, Future[Done]] = buildAllTvSeriesSink(logger = logger)
 
     val startingTime: Long = System.currentTimeMillis()
     //graph sink->flow->sink
-    source
+    val results: Future[Done] = source
       .via(flow = buildTvSerieFlow())
       .runWith(sink = sink)
       .andThen {
@@ -197,7 +255,10 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
           logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
         case Failure(error: Error) => logger.error(s"$error")
       }
+
+    results
   }
+
 }