first commit

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

@@ -36,21 +36,21 @@ public class MovieController {
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 
-    @RequestMapping(value = "/movies", method = RequestMethod.GET)
-    private ResponseEntity<String> getAllMovies() {
+    @RequestMapping(value = "/tvseries", method = RequestMethod.GET)
+    private ResponseEntity<String> getAllvSeries() {
         akkaStreamFilesProcessing.getAllTvSeries();
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 
-    @RequestMapping(value = "/movies/{movieTitle}", method = RequestMethod.GET)
-    private ResponseEntity<String> getMovieByTitle(@PathVariable(name ="movieTitle" ) String movieTitle){
-        akkaStreamFilesProcessing.getByTvSeriePrimaryTitle(movieTitle);
-        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 = "/movies/title/{tvSerieTitle}", method = RequestMethod.GET)
+    @RequestMapping(value = "/tvseries/title/{tvSerieTitle}", method = RequestMethod.GET)
     private ResponseEntity<String> getListOfPersonsIDByTvSerieID(@PathVariable(name = "tvSerieTitle") String tvSerieTitle){
-        akkaStreamFilesProcessing.generalize(tvSerieTitle);
+      akkaStreamFilesProcessing.getPersonsTeamForTvSerie(tvSerieTitle);
        return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 }

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

@@ -16,16 +16,9 @@ object Models {
         ", birth-year:" + birthYear +
         ", dearth year:" + deathYear +
         ", primary profession:" + primaryProfession +
-        ", known for titles:" + knownForTitles +
+      ", known-for-titles"+knownForTitles+
         "]"
     }
-
-    def getNconst: String={
-      nconst
-    }
-    def getPrimaryName: String = {
-      primaryName
-    }
   }
 
   final case class TvSeries(
@@ -33,7 +26,6 @@ object Models {
                              titleType: String,
                              primaryTitle: String,
                              originalTitle: String,
-                             isAdult: String,
                              startYear: String,
                              endYear: String,
                              runtimeMinutes: String,
@@ -45,18 +37,10 @@ object Models {
         ", title-type:" + titleType +
         ", primary-title:" + primaryTitle + "" +
         ", riginal-title:" + originalTitle +
-        ", is adult (no=0, yes=1):" + isAdult +
         ", start year:" + startYear +
         ", end year:" + endYear +
         ", runtime minutes:" + runtimeMinutes +
-        ", genre:" + genres + "]"
-    }
-
-   def getTconst: String= {
-     tconst
-   }
-    def getPrimaryTitle: String={
-      primaryTitle
+        ", genres:" + genres + "]"
     }
   }
 

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

@@ -2,28 +2,27 @@ package fr.natan.akkastreamfileprocessingapi.models
 import Models.{Person, TvSeries}
 object ModelsBuilder {
 
-  def buildPersonModel(map: Map[String, String]): Person = {
+  def buildPersonModel(personMap: Map[String, String]): Person = {
     Person(
-      map("nconst"),
-      map("primaryName"),
-      map("birthYear"),
-      map("deathYear"),
-      map("primaryProfession").split(",").toList,
-      map("knownForTitles").split(",").toList,
+      personMap("nconst"),
+      personMap("primaryName"),
+      personMap("birthYear"),
+      personMap("deathYear"),
+      personMap("primaryProfession").split(",").toList,
+      personMap("knownForTitles").split(",").toList
     )
   }
 
-  def buildTvSerieModel(map: Map[String, String]): TvSeries = {
+  def buildTvSerieModel(tvSerieMap: Map[String, String]): TvSeries = {
     val tvSerie: TvSeries = TvSeries(
-      map("tconst"),
-      map("titleType"),
-      map("primaryTitle"),
-      map("originalTitle"),
-      map("isAdult"),
-      map("startYear"),
-      map("endYear"),
-      map("runtimeMinutes"),
-      map("genres")
+      tvSerieMap("tconst"),
+      tvSerieMap("titleType"),
+      tvSerieMap("primaryTitle"),
+      tvSerieMap("originalTitle"),
+      tvSerieMap("startYear"),
+      tvSerieMap("endYear"),
+      tvSerieMap("runtimeMinutes"),
+      tvSerieMap("genres")
     )
     tvSerie
   }

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

@@ -48,10 +48,10 @@ object AkkaStreamComponents {
   def filterByTvSeriePrimaryTitleFlow(tvSeriePrimaryTitle: String): Flow[Map[String, String], TvSeries, NotUsed] = {
     val filterFlow: Flow[Map[String, String], TvSeries, NotUsed] = Flow[Map[String, String]]
         .filter((rows: Map[String, String]) => {
-          rows.getOrElse(key = "primaryName", default = "")==tvSeriePrimaryTitle
+          rows.getOrElse(key = "primaryTitle", default = "")==tvSeriePrimaryTitle
         })
         .map(rowMap => {
-          val tvSerie: TvSeries = buildTvSerieModel(map = rowMap)
+          val tvSerie: TvSeries = buildTvSerieModel(tvSerieMap = rowMap)
           tvSerie
         })
 
@@ -66,7 +66,7 @@ object AkkaStreamComponents {
           rowMap.getOrElse(key="nconst",default="")==nconst
         })
         .map(rowMap=>{
-          buildPersonModel(map = rowMap)
+          buildPersonModel(personMap = rowMap)
         })
 
     personFilter
@@ -80,7 +80,7 @@ object AkkaStreamComponents {
           rowMap.getOrElse(key = "primaryName", default = "")==personName
         })
         .map((rowMap: Map[String, String])=>{
-          buildPersonModel(map = rowMap)
+          buildPersonModel(personMap = rowMap)
         })
 
     personFilter
@@ -103,7 +103,7 @@ object AkkaStreamComponents {
       }
       )
       .via(Compression.gunzip())
-      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
+      .via(CsvParsing.lineScanner(CsvParsing.Tab))
       .via(CsvToMap.toMapAsStrings())
 
     datasource
@@ -121,18 +121,20 @@ object AkkaStreamComponents {
   //sinks building
   //sink1
   def buildTvSeriesSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
-    val tvSeriesSink : Sink[TvSeries, Future[Done]]= Sink.foreach[TvSeries](
-      (tvSerie: TvSeries) => {
-        logger.info(s"${tvSerie.toString}")
-      }
-    )
+    val tvSeriesSink : Sink[TvSeries, Future[Done]]= Sink
+      .foreach((tvSerie: TvSeries)=> logger.info(s"${tvSerie.toString}"))
     tvSeriesSink
   }
 
+  def buildAllTvSeriesSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
+    val tvSeriesSink: Sink[TvSeries, Future[Done]] = Sink
+      .foreach((tvSerie: TvSeries)=>logger.info(s"${tvSerie.toString}"))
+    tvSeriesSink
+  }
   //sink2
   def buildAllPersonsSink(logger: Logger): Sink[Person,Future[Done]] = {
     val listPersonsSink: Sink[Person, Future[Done]]=
-      Sink.foreach[Person]((person: Person)=>logger.info(s"${person.toString}"))
+      Sink.foreach((person: Person)=>logger.info(s"${person.toString}"))
     listPersonsSink
   }
 

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

@@ -1,18 +1,21 @@
 package fr.natan.akkastreamfileprocessingapi.service
 
-import fr.natan.akkastreamfileprocessingapi.models.Models.Person
+import akka.stream.scaladsl.Source
+import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
+
+import scala.concurrent.Future
 
 trait AkkaStreamFileProcessing {
   def getAllPersons(): Unit
 
-  def getPersonById(nconst: String): Person
+  def getPersonById(nconst: String): Future[Person]
   def getPersonByName(primaryName: String): Unit
 
   def getAllTvSeries(): Unit
 
-  def getByTvSeriePrimaryTitle(tvSerieTitle: String): Unit
+  def getByTvSeriePrimaryTitle(tvSerieTitle: String):Future[Seq[TvSeries]]
 
-  def generalize(tvSeriePrimaryTitle: String): Unit
+  def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String)
 
 
 }

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

@@ -7,18 +7,7 @@ import com.typesafe.scalalogging.slf4j.Logger
 import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics, titlePrincipalsBasics}
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
 import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.{buildPersonModel, buildTvSerieModel}
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
-  buildAllPersonsSink,
-  buildAndValidateSource,
-  buildPersonFlow,
-  buildPersonSink,
-  buildSource,
-  buildTvSerieFlow,
-  buildTvSeriesSink,
-  filterByPersonIdFlow,
-  filterByPersonNameFlow,
-  filterByTvSeriePrimaryTitleFlow
-}
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAllPersonsSink, buildAllTvSeriesSink, buildAndValidateSource, buildPersonFlow, buildPersonSink, buildSource, buildTvSerieFlow, buildTvSeriesSink, filterByPersonIdFlow, filterByPersonNameFlow, filterByTvSeriePrimaryTitleFlow}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -34,41 +23,37 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
   implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getAllPersons(): Unit = {
+  override def getAllPersons() = {
     val personSource: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
-    val personSink: Sink[Person, Future[Done]] = buildAllPersonsSink(logger = logger)
-
     //graph
     val startTime: Long = System.currentTimeMillis()
     val result: Future[Done] = personSource
       .via(flow = buildPersonFlow())
-      .runWith(sink = personSink)
-      .andThen {
-        case Success(value) =>
-          val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
-          logger.info(s"$value: Successfully processed, elapsed time: $elapsedTime")
-        case Failure(exception) => logger.error(s"$exception: Fail")
-      }
+      .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): Person = {
+  override def getPersonById(nconst: String): Future[Person] = {
     val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
 
     val startTime: Long = System.currentTimeMillis()
     val res = source
       .via(flow = filterByPersonIdFlow(nconst = nconst))
       .runWith(sink = buildPersonSink())
-     .andThen {
-        case Success(value) =>
-          val elapsedTime: Long = (System.currentTimeMillis()-startTime)/1000
-          logger.info(s"$value")
-          logger.info(s"Successfully processed, elapsed time: $elapsedTime")
-        case Failure(exception) => logger.error(s"$exception: Fail")
-      }
 
-    Await.result(res,1 minutes)
-    val person: Person = res.value.get.get
-    person
+    res.andThen {
+      case Failure(exception) => logger.info(s"error!!!!!!!!!!!!!$exception")
+      case Success(value) => logger.info(s"$value")
+    }
+
+    res
   }
 
   override def getPersonByName(primaryName: String): Unit = {
@@ -87,7 +72,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
   override def getAllTvSeries(): Unit = {
 
     val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
-    val sink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
+    val sink: Sink[TvSeries, Future[Done]] = buildAllTvSeriesSink(logger = logger)
 
     val startingTime: Long = System.currentTimeMillis()
 
@@ -104,26 +89,21 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
   }
 
-  override def getByTvSeriePrimaryTitle(tvSeriePrimaryTitle: String): Unit = {
+  override def getByTvSeriePrimaryTitle(tvSeriePrimaryTitle: String):Future[IndexedSeq[TvSeries]] = {
 
     val tvSeriesSource: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
-    val tvSeriesSink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
 
-    val filterByMovieTitleFlow: Flow[Map[String, String], TvSeries, NotUsed] =
+    val filterByTvSerieTitle: Flow[Map[String, String], TvSeries, NotUsed] =
       filterByTvSeriePrimaryTitleFlow(tvSeriePrimaryTitle = tvSeriePrimaryTitle)
 
-    val startTime: Long = System.currentTimeMillis()
-    val listTvSeries: Future[Done] = tvSeriesSource
-      .via(flow = filterByMovieTitleFlow)
-      .runWith(sink = tvSeriesSink)
-      .andThen {
-        case Success(value) =>
-          val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
-          logger.info(s"$value: successfully processing file, elapsed time: $elapsedTime sec")
-        case Failure(error: Error) => logger.error(s"$error")
-      }
+    val tvSrie :Future[IndexedSeq[TvSeries]] = tvSeriesSource
+      .via(flow = filterByTvSerieTitle)
+      .runWith(Sink.collection)
 
-    getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
+      tvSrie.andThen {
+      case Failure(exception) => logger.info(s"$exception")
+      case Success(value) => logger.info(s"$value")
+    }
 
   }
 
@@ -193,7 +173,6 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
         case Failure(exception) => logger.error(s"$exception")
         case Success(value) =>
           val elapsedTime: Long = (System.currentTimeMillis()-startTime)/1000
-          logger.info(s"$value")
           logger.info(s"END: Successfully, elapsed time: $elapsedTime")
       }
 
@@ -203,12 +182,17 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     listPersons
 
   }
-
-  override def generalize(tvSeriePrimaryTitle: String): Unit={
+  override def getPersonsTeamForTvSerie(tvSeriePrimaryTitle: String):Unit={
 
     val tvSerieID: String = getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
     val listPersonIDs: List[Option[String]] = getListOfPersonsIDByTvSerieID(tconst = tvSerieID)
-    getListOfPersonsForTvSerie(nconstList = listPersonIDs)
+    val personsTeam: List[Person] = getListOfPersonsForTvSerie(nconstList = listPersonIDs)
+
+    logger.info(s"Team size:${personsTeam.size}")
+    personsTeam.foreach((person: Person)=>{
+      logger.info(s"${person.toString}")
+    })
+
   }
 
 }