first commit

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

@@ -8,6 +8,9 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.Collection;
+import java.util.List;
+
 @RestController
 public class MovieController {
 
@@ -47,4 +50,10 @@ public class MovieController {
         akkaStreamFilesProcessing.getByTvSeriePrimaryTitle(movieTitle);
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
+
+    @RequestMapping(value = "/movies/id/{tvSerieID}", method = RequestMethod.GET)
+    private ResponseEntity<String> getListOfPersonsIDByTvSerieID(@PathVariable(name = "tvSerieID") String tvSerieID){
+       akkaStreamFilesProcessing.getListOfPersonsIDByTvSerieID(tvSerieID);
+       return new ResponseEntity<>("is running", HttpStatus.OK);
+    }
 }

src/main/scala/fr/natan/akkastreamfileprocessingapi/datasource/Datasource.scala

@@ -6,6 +6,6 @@ object Datasource {
 
   val titleBasics: File = new File("../title.basics.tsv.gz")
   val nameBasics: File = new File("../name.basics.tsv.gz")
-  val titlePrincipalsBasics: File = new File("../title.basics.tsv.gz")
+  val titlePrincipalsBasics: File = new File("../title.principals.tsv.gz")
   val separator: String = "\t"
 }

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

@@ -12,6 +12,7 @@ import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.{buildPersonMod
 
 import java.io.File
 import java.nio.file.Paths
+import java.util
 import scala.concurrent.Future
 
 object AkkaStreamComponents {
@@ -102,7 +103,7 @@ object AkkaStreamComponents {
       }
       )
       .via(Compression.gunzip())
-      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote))
+      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
       .via(CsvToMap.toMapAsStrings())
 
     datasource

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

@@ -11,4 +11,6 @@ trait AkkaStreamFileProcessing {
   def getAllTvSeries(): Unit
 
   def getByTvSeriePrimaryTitle(tvSerieTitle: String): Unit
+
+  def getListOfPersonsIDByTvSerieID(tconst: String): List[Option[String]]
 }

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

@@ -1,10 +1,12 @@
 package fr.natan.akkastreamfileprocessingapi.service
 
 import akka.actor.ActorSystem
+import akka.stream.ActorAttributes.supervisionStrategy
+import akka.stream.Supervision.resumingDecider
 import akka.stream.scaladsl.{Flow, Sink, Source}
 import akka.{Done, NotUsed}
 import com.typesafe.scalalogging.slf4j.Logger
-import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics}
+import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
 import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildTvSerieModel
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAllPersonsSink, buildAndValidateSource, buildPersonFlow, buildPersonSink, buildSource, buildTvSerieFlow, buildTvSeriesSink, filterByPersonIdFlow, filterByPersonNameFlow, filterByTvSeriePrimaryTitleFlow}
@@ -14,7 +16,9 @@ import org.springframework.stereotype.Component
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.DurationInt
 import scala.concurrent.{Await, Future}
+import scala.language.postfixOps
 import scala.util.{Failure, Success}
+import scala.collection.immutable.Iterable
 
 @Component
 class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
@@ -49,7 +53,8 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
      .andThen {
         case Success(value) =>
           val elapsedTime: Long = (System.currentTimeMillis()-startTime)/1000
-          logger.info(s"$value: Successfully processed, elapsed time: $elapsedTime")
+          logger.info(s"$value")
+          logger.info(s"Successfully processed, elapsed time: $elapsedTime")
         case Failure(exception) => logger.error(s"$exception: Fail")
       }
 
@@ -111,27 +116,59 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
       }
 
     getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
+
   }
 
-  private def getTvSerieIdByPrimaryTitle(primaryTitle: String): String ={
-    val source: Source[Map[String, String],_]= buildSource(inputFile = titleBasics)
+  private def getTvSerieIdByPrimaryTitle(primaryTitle: String): String = {
+    val source: Source[Map[String, String], _] = buildSource(inputFile = titleBasics)
 
-    var tvSerie: TvSeries =null
-    val res : Future[Option[String]]= source
-      .filter((rowMap: Map[String, String]) =>{
-        rowMap.getOrElse(key = "primaryTitle","")==primaryTitle
+    var tvSerie: TvSeries = null
+    val res: Future[Option[String]] = source
+      .filter((rowMap: Map[String, String]) => {
+        rowMap.getOrElse(key = "primaryTitle", "") == primaryTitle
       })
-      .map((rowMap: Map[String, String])=>{
+      .map((rowMap: Map[String, String]) => {
         tvSerie = buildTvSerieModel(rowMap)
         rowMap.get("tconst")
       })
       .runWith(Sink.head)
 
-   Await.result(res,1 minutes)
-    val tconst:String = res.value.get.get.get
+    Await.result(res, 1 minutes)
+    val tconst: String = res.value.get.get.get
+    logger.info(s"TvSerie ID: $tconst, $tvSerie")
     tconst
   }
 
+  def getListOfPersonsIDByTvSerieID(tconst: String): List[Option[String]]={
+    val source: Source[Map[String, String], _] = buildSource(inputFile = titlePrincipalsBasics)
+    var tvSeries: Iterable[TvSeries]=null
+
+    val startTime : Long = System.currentTimeMillis()
+    val res: Future[IndexedSeq[Option[String]]] = source
+      .filter((rowMaps: Map[String, String])=>{
+        rowMaps.getOrElse(key = "tconst", default = "")==tconst
+      })
+      .map((rowMap: Map[String, String])=>{
+        rowMap.get(key = "nconst")
+      })
+      .withAttributes(supervisionStrategy(resumingDecider))
+      .runWith(Sink.collection)
+      .andThen {
+        case Success(value) =>
+          logger.info(s"$value")
+          val elapsedTime : Long = (System.currentTimeMillis() - startTime)/1000
+          logger.info(s"success, elapsed time: $elapsedTime")
+
+        case Failure(error: Error) => logger.error(s"$error")
+      }
+
+    Await.result(res,5 minutes)
+    val listPersonsID: List[Option[String]] = res.value.get.get.toList
+
+    listPersonsID
+
+  }
+
 
 }