first commit

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

@@ -1,5 +1,6 @@
 package fr.natan.akkastreamfileprocessingapi.controller;
 
+import fr.natan.akkastreamfileprocessingapi.models.Models;
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamFileProcessing;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -7,6 +8,8 @@ import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
+import scala.collection.IndexedSeq;
+import scala.concurrent.Future;
 
 @SuppressWarnings("SpellCheckingInspection")
 @RestController
@@ -18,22 +21,31 @@ public class MovieController {
         this.akkaStreamFilesProcessing = akkaStreamFilesProcessing;
     }
 
-    @RequestMapping(value = "/persons", method = RequestMethod.GET)
-    private ResponseEntity<String> getAllPersons(){
-        akkaStreamFilesProcessing.getAllPersons();
-        return new ResponseEntity<>("is running", HttpStatus.OK);
-    }
-
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
-    private ResponseEntity<String> getPersonByID(@PathVariable(name = "personID") String nconst){
+    private Future<Models.Person> getPersonByID(@PathVariable(name = "personID") String nconst){
 
-       return new ResponseEntity<>(akkaStreamFilesProcessing.getPersonById(nconst).toString(), HttpStatus.OK);
+       return akkaStreamFilesProcessing.getPersonById(nconst);
 
     }
 
     @RequestMapping(value = "/persons/name/{primaryName}", method = RequestMethod.GET)
-    private ResponseEntity<String> getPersonByName(@PathVariable(name = "primaryName") String primaryName){
-        akkaStreamFilesProcessing.getPersonByName(primaryName);
+    private Future<IndexedSeq<Models.Person>> getPersonByName(@PathVariable(name = "primaryName") String primaryName){
+        return akkaStreamFilesProcessing.getPersonByName(primaryName);
+    }
+
+    @RequestMapping(value = "/tvseries/{tvseriePrimaryTitle}", method = RequestMethod.GET)
+    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.getPersonsTeamForTvSerie(tvSerieTitle);
+    }
+
+    @RequestMapping(value = "/persons", method = RequestMethod.GET)
+    private ResponseEntity<String> getAllPersons(){
+        akkaStreamFilesProcessing.getAllPersons();
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 
@@ -42,16 +54,4 @@ public class MovieController {
         akkaStreamFilesProcessing.getAllTvSeries();
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
-
-    @RequestMapping(value = "/tvseries/{tvseriePrimaryTitle}", method = RequestMethod.GET)
-    private ResponseEntity<String> getTvserieByPrimaryTitle(@PathVariable(name ="tvseriePrimaryTitle" ) String tvseriePrimaryTitle){
-        akkaStreamFilesProcessing.getByTvSeriePrimaryTitle(tvseriePrimaryTitle);
-        return new ResponseEntity<>("OK", HttpStatus.OK);
-    }
-
-    @RequestMapping(value = "/tvseries/title/{tvSerieTitle}", method = RequestMethod.GET)
-    private ResponseEntity<String> getListOfPersonsForTvSerie(@PathVariable(name = "tvSerieTitle") String tvSerieTitle){
-      akkaStreamFilesProcessing.getPersonsTeamForTvSerie(tvSerieTitle);
-       return new ResponseEntity<>("is running", HttpStatus.OK);
-    }
 }

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

@@ -9,9 +9,9 @@ trait AkkaStreamFileProcessing {
   def getPersonById(nconst: String): Future[Person]
   def getPersonByName(primaryName: String): Future[IndexedSeq[Person]]
 
-  def getByTvSeriePrimaryTitle(tvSerieTitle: String):Future[Seq[TvSeries]]
+  def getTvSerieByPrimaryTitle(tvSerieTitle: String):Future[IndexedSeq[TvSeries]]
 
-  def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String): Future[List[Person]]
+  def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]
   def getAllTvSeries(): Unit
   def getAllPersons(): Unit
 }

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

@@ -8,8 +8,9 @@ import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, t
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
 import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildPersonModel
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
-  buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow,
-  buildSource, buildTvSerieFlow, filterByPersonIdFlow, filterByPersonNameFlow, filterByTvSeriePrimaryTitleFlow}
+  buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildSource, buildTvSerieFlow,
+  filterByPersonIdFlow, filterByPersonNameFlow, filterByTvSeriePrimaryTitleFlow
+}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -25,23 +26,6 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
   implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getAllPersons(): Unit = {
-    val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    //graph
-    val startTime: Long = System.currentTimeMillis()
-    val result: Future[Done] = personSource
-      .via(flow = buildPersonFlow())
-      .runWith(sink = buildAllPersonsSink(logger = logger))
-
-    result.onComplete {
-      case Failure(exception) => logger.error(s"$exception")
-      case Success(value) =>
-        logger.info(s"$value")
-        val time: Long = (System.currentTimeMillis() - startTime) / 100
-        logger.info(s"elapsed time: $time")
-    }
-  }
-
   override def getPersonById(nconst: String): Future[Person] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
     val res = source
@@ -62,44 +46,35 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
       .via(flow = filterByPersonNameFlow(primaryName = primaryName))
       .runWith(Sink.collection)
 
-    persons
-  }
-
-  override def getAllTvSeries(): Unit = {
-
-    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
-    val sink: Sink[TvSeries, Future[Done]] = buildAllTvSeriesSink(logger = logger)
-
-    val startingTime: Long = System.currentTimeMillis()
-
-    //graph sink->flow->sink
-    source
-      .via(flow = buildTvSerieFlow())
-      .runWith(sink = sink)
-      .andThen {
-        case Success(value) =>
-          val elapsedTime: Long = (System.currentTimeMillis() - startingTime) / 1000
-          logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
-        case Failure(error: Error) => logger.error(s"$error")
-      }
+    persons.onComplete({
+      case Failure(exception) => logger.error(s"$exception")
+      case Success(value: IndexedSeq[Person]) =>
+        value.foreach((person: Person)=>logger.info(s"$person"))
+        logger.info("SUCCESS")
+    })
 
+    persons
   }
 
-  override def getByTvSeriePrimaryTitle(tvSeriePrimaryTitle: String):Future[IndexedSeq[TvSeries]] = {
+  override def getTvSerieByPrimaryTitle(tvSeriePrimaryTitle: String):Future[IndexedSeq[TvSeries]] = {
 
     val tvSeriesSource: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
 
     val filterByTvSerieTitle: Flow[Map[String, String], TvSeries, NotUsed] =
       filterByTvSeriePrimaryTitleFlow(tvSeriePrimaryTitle = tvSeriePrimaryTitle)
 
-    val tvSrie :Future[IndexedSeq[TvSeries]] = tvSeriesSource
+    val tvSries :Future[IndexedSeq[TvSeries]] = tvSeriesSource
       .via(flow = filterByTvSerieTitle)
       .runWith(Sink.collection)
 
-      tvSrie.andThen {
+      tvSries.onComplete({
       case Failure(exception) => logger.info(s"$exception")
-      case Success(value) => logger.info(s"$value")
-    }
+      case Success(value: IndexedSeq[TvSeries]) =>
+        value.foreach((tvSrie: TvSeries)=>logger.info(s"$tvSrie"))
+        logger.info("SUCCESS")
+      })
+
+    tvSries
   }
 
   private def getTvSerieIdByPrimaryTitle(primaryTitle: String): Future[Option[String]] = {
@@ -115,10 +90,9 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     res
   }
 
-  private def getListOfPersonsIDByTvSerieID(tvSerieID: Future[Option[String]]): Future[List[Option[String]]]={
+  private def getListOfPersonsIDByTvSerieID(tvSerieID: Future[Option[String]]): Future[IndexedSeq[Option[String]]]={
     val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
-
-    val res: Future[List[Option[String]]] = source
+    val res: Future[IndexedSeq[Option[String]]] = source
       .filter((rowMaps: Map[String, String])=>{
         rowMaps.getOrElse(key = "tconst", default = "")==tvSerieID.value.get.get.get
       })
@@ -126,13 +100,14 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
         rowMap.get(key = "nconst")
       })
       .runWith(Sink.collection)
+
     res
   }
 
-  private def getListOfPersonsForTvSerie(listPersonsIDs: Future[List[Option[String]]]): Future[List[Person]] = {
+  private def getListOfPersonsForTvSerie(listPersonsIDs: Future[IndexedSeq[Option[String]]]): Future[IndexedSeq[Person]] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
 
-    val res : Future[List[Person]]=
+    val res : Future[IndexedSeq[Person]]=
       source
         .filter((rowMaps: Map[String, String])=>{
           listPersonsIDs.value.get.get.contains(rowMaps.get(key = "nconst"))
@@ -144,10 +119,10 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
     res
   }
-  override def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String):Future[List[Person]]={
+  override def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String):Future[IndexedSeq[Person]]={
 
     logger.info("STEP 1/3")
-    val tvSerieID = getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
+    val tvSerieID : Future[Option[String]]= getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
     tvSerieID.andThen({
       case Failure(exception) => logger.error(s"!!!!!!!!!!!!!!!!!!!!1$exception")
       case Success(value: Option[String]) =>
@@ -155,16 +130,16 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
         logger.info("END STEP 1/3")
     })
     logger.info("STEP 2/3")
-    val listPersonIDs: Future[List[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieID = tvSerieID)
+    val listPersonIDs: Future[IndexedSeq[Option[String]]] = getListOfPersonsIDByTvSerieID(tvSerieID = tvSerieID)
     listPersonIDs.andThen({
       case Failure(exception) => logger.error(s"!!!!!!!!!!!!!!!!!!!!!2$exception")
-      case Success(value: List[Option[String]]) =>
+      case Success(value) =>
         value.foreach((personID:Option[String])=>logger.info(s"Person ID:$personID"))
         logger.info("END STEP 2/3")
     })
 
     logger.info("STEP 3/3")
-    val personsTeam: Future[List[Person]] = getListOfPersonsForTvSerie(listPersonsIDs = listPersonIDs)
+    val personsTeam: Future[IndexedSeq[Person]] = getListOfPersonsForTvSerie(listPersonsIDs = listPersonIDs)
     personsTeam.andThen({
       case Failure(exception) => logger.error(s"!!!!!!!!!!!!!!!!!!!!!!!3$exception")
       case Success(value) =>
@@ -175,6 +150,41 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
    null
   }
+
+  override def getAllPersons(): Unit = {
+    val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+    //graph
+    val startTime: Long = System.currentTimeMillis()
+    val result: Future[Done] = personSource
+      .via(flow = buildPersonFlow())
+      .runWith(sink = buildAllPersonsSink(logger = logger))
+
+    result.onComplete {
+      case Failure(exception) => logger.error(s"$exception")
+      case Success(value) =>
+        logger.info(s"$value")
+        val time: Long = (System.currentTimeMillis() - startTime) / 100
+        logger.info(s"elapsed time: $time")
+    }
+  }
+
+  override def getAllTvSeries(): Unit = {
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
+    val sink: Sink[TvSeries, Future[Done]] = buildAllTvSeriesSink(logger = logger)
+
+    val startingTime: Long = System.currentTimeMillis()
+    //graph sink->flow->sink
+    source
+      .via(flow = buildTvSerieFlow())
+      .runWith(sink = sink)
+      .andThen {
+        case Success(value) =>
+          val elapsedTime: Long = (System.currentTimeMillis() - startingTime) / 1000
+          logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
+        case Failure(error: Error) => logger.error(s"$error")
+      }
+  }
+
 }
 
 

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

@@ -1,8 +0,0 @@
-package fr.natan.akkastreamfileprocessingapi.service
-
-import org.springframework.stereotype.Component
-
-@Component
-class CompleteFuture {
-
-}