first commit

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

@@ -17,18 +17,6 @@ public class MovieController {
         this.akkaStreamFilesProcessing = akkaStreamFilesProcessing;
     }
 
-    @RequestMapping(value = "/movies", method = RequestMethod.GET)
-    private ResponseEntity<String> getAllMovies() {
-        akkaStreamFilesProcessing.getAllMovies();
-        return new ResponseEntity<>("is running", HttpStatus.OK);
-    }
-
-    @RequestMapping(value = "/movies/{movieTitle}", method = RequestMethod.GET)
-    private ResponseEntity<String> getMovieByTitle(@PathVariable(name ="movieTitle" ) String movieTitle){
-        akkaStreamFilesProcessing.getMoviesByTitle(movieTitle);
-        return new ResponseEntity<>("is running", HttpStatus.OK);
-    }
-
     @RequestMapping(value = "/persons", method = RequestMethod.GET)
     private ResponseEntity<String> getallPersons(){
         akkaStreamFilesProcessing.getAllPersons();
@@ -37,9 +25,8 @@ public class MovieController {
 
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
     private ResponseEntity<String> getPersonByID(@PathVariable(name = "personID") String nconst){
-       akkaStreamFilesProcessing.getPersonById(nconst);
 
-       return new ResponseEntity<>("is running", HttpStatus.OK);
+       return new ResponseEntity<>(akkaStreamFilesProcessing.getPersonById(nconst).toString(), HttpStatus.OK);
 
     }
 
@@ -48,4 +35,16 @@ public class MovieController {
         akkaStreamFilesProcessing.getPersonByName(primaryName);
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
+
+    @RequestMapping(value = "/movies", method = RequestMethod.GET)
+    private ResponseEntity<String> getAllMovies() {
+        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);
+    }
 }

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

@@ -19,6 +19,13 @@ object Models {
         ", known for titles:" + knownForTitles +
         "]"
     }
+
+    def getNconst: String={
+      nconst
+    }
+    def getPrimaryName: String = {
+      primaryName
+    }
   }
 
   final case class TvSeries(
@@ -44,5 +51,13 @@ object Models {
         ", runtime minutes:" + runtimeMinutes +
         ", genre:" + genres + "]"
     }
+
+   def getTconst: String= {
+     tconst
+   }
+    def getPrimaryTitle: String={
+      primaryTitle
+    }
   }
+
 }

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

@@ -21,17 +21,6 @@ object AkkaStreamComponents {
   //flows building
 
   //flow1
-  def buildTvSerieFlow(): Flow[Map[String, String], TvSeries, NotUsed] = {
-
-    val tvFlow: Flow[Map[String, String], TvSeries, NotUsed] =
-      Flow[Map[String, String]]
-        .map(row => {
-          buildTvSerieModel(row)
-        })
-    tvFlow
-  }
-
-  //flow2
   def buildPersonFlow(): Flow[Map[String, String], Person, NotUsed] = {
     val personFlow: Flow[Map[String, String], Person, NotUsed] =
       Flow[Map[String, String]]
@@ -42,14 +31,27 @@ object AkkaStreamComponents {
     personFlow
   }
 
+  //flow2
+
+  def buildTvSerieFlow(): Flow[Map[String, String], TvSeries, NotUsed] = {
+
+    val tvFlow: Flow[Map[String, String], TvSeries, NotUsed] =
+      Flow[Map[String, String]]
+        .map(row => {
+          buildTvSerieModel(row)
+        })
+    tvFlow
+  }
+
   //flow3
-  def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: String): Flow[Map[String, String], TvSeries, NotUsed] = {
+  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("primaryTitle","")==moviePrimaryTitle
+          rows.getOrElse(key = "primaryName", default = "")==tvSeriePrimaryTitle
         })
         .map(rowMap => {
-          buildTvSerieModel(map = rowMap)
+          val tvSerie: TvSeries = buildTvSerieModel(map = rowMap)
+          tvSerie
         })
 
     filterFlow
@@ -60,7 +62,7 @@ object AkkaStreamComponents {
     val personFilter: Flow[Map[String, String], Person, NotUsed]=
       Flow[Map[String, String]]
         .filter((rowMap:Map[String, String])=>{
-          rowMap.getOrElse("nconst","")==nconst
+          rowMap.getOrElse(key="nconst",default="")==nconst
         })
         .map(rowMap=>{
           buildPersonModel(map = rowMap)
@@ -69,11 +71,12 @@ object AkkaStreamComponents {
     personFilter
   }
 
+  //flow5
   def filterByPersonNameFlow(personName: String): Flow[Map[String, String], Person, NotUsed] ={
     val personFilter: Flow[Map[String, String], Person, NotUsed] =
       Flow[Map[String, String]]
         .filter((rowMap: Map[String, String]) =>{
-          rowMap.getOrElse("primaryName","")==personName
+          rowMap.getOrElse(key = "primaryName", default = "")==personName
         })
         .map((rowMap: Map[String, String])=>{
           buildPersonModel(map = rowMap)
@@ -83,10 +86,10 @@ object AkkaStreamComponents {
   }
 
   //source building
-  def buildSource(inputFile: File): Source[Map[String, String], NotUsed] = {
 
-    var datasource: Source[Map[String, String], NotUsed] = null
+  def buildSource(inputFile: File): Source[Map[String, String], _] = {
 
+    var datasource: Source[Map[String, String], NotUsed] = null
     if (!fileExists(inputFile.getPath)) {
       return null
     }
@@ -105,9 +108,9 @@ object AkkaStreamComponents {
     datasource
   }
 
-  def buildAndValidateSource(inputFile: File): Source[Map[String, String], NotUsed] = {
+  def buildAndValidateSource(inputFile: File): Source[Map[String, String], _] = {
 
-    val source: Source[Map[String, String], NotUsed] = buildSource(inputFile = inputFile)
+    val source: Source[Map[String, String], _] = buildSource(inputFile = inputFile)
     if (source == null) {
       throw new FileNotFoundException(filename = inputFile.getPath)
     }
@@ -115,28 +118,25 @@ object AkkaStreamComponents {
   }
 
   //sinks building
-
   //sink1
   def buildTvSeriesSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
     val tvSeriesSink : Sink[TvSeries, Future[Done]]= Sink.foreach[TvSeries](
-      (movie: TvSeries) => {
-        logger.info(s"${movie.toString}")
+      (tvSerie: TvSeries) => {
+        logger.info(s"${tvSerie.toString}")
       }
     )
     tvSeriesSink
   }
 
   //sink2
-  def buildAllPersonsSink(logger: Logger): Sink[Person,Future[Seq[Person]]] = {
-    val listPersonsSink: Sink[Person, Future[Seq[Person]]]=
-      Sink.seq
+  def buildAllPersonsSink(logger: Logger): Sink[Person,Future[Done]] = {
+    val listPersonsSink: Sink[Person, Future[Done]]=
+      Sink.foreach[Person]((person: Person)=>logger.info(s"${person.toString}"))
     listPersonsSink
   }
 
   //sink3
-  def buildPersonSink(logger: Logger): Sink[Person, Future[Done]] = {
-    Sink.foreach[Person](
-      (person: Person) => logger.info(s"${person.toString}")
-    )
+  def buildPersonSink(): Sink[Person, Future[Person]] = {
+    Sink.head
   }
 }

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

@@ -1,12 +1,14 @@
 package fr.natan.akkastreamfileprocessingapi.service
 
+import fr.natan.akkastreamfileprocessingapi.models.Models.Person
+
 trait AkkaStreamFileProcessing {
+  def getAllPersons(): Unit
 
-  def getAllMovies()
-  def getMoviesByTitle (movieTitle: String)
+  def getPersonById(nconst: String): Person
+  def getPersonByName(primaryName: String): Unit
 
-  def getAllPersons()
+  def getAllTvSeries(): Unit
 
-  def getPersonById(nconst: String)
-  def getPersonByName(primaryName: String)
+  def getByTvSeriePrimaryTitle(tvSerieTitle: String): Unit
 }

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

@@ -6,12 +6,14 @@ import akka.{Done, NotUsed}
 import com.typesafe.scalalogging.slf4j.Logger
 import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics}
 import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildPersonFlow, buildPersonSink, buildSource, buildTvSerieFlow, buildTvSeriesSink, filterByMoviePrimaryTitleFlow, filterByPersonIdFlow, filterByPersonNameFlow}
+import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.buildTvSerieModel
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAllPersonsSink, buildAndValidateSource, buildPersonFlow, buildPersonSink, buildSource, buildTvSerieFlow, buildTvSeriesSink, filterByPersonIdFlow, filterByPersonNameFlow, filterByTvSeriePrimaryTitleFlow}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
 import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Future
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.{Await, Future}
 import scala.util.{Failure, Success}
 
 @Component
@@ -20,9 +22,58 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
   implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getAllMovies(): Unit = {
+  override def getAllPersons(): Unit = {
+    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")
+      }
+  }
+
+  override def getPersonById(nconst: String): 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: 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
+  }
+
+  override def getPersonByName(primaryName: String): Unit = {
+    val source: Source[Map[String, String], _] = buildSource(inputFile = nameBasics)
+    source
+      .via(
+        flow = filterByPersonNameFlow(personName = primaryName)
+      )
+      .runWith(sink = buildPersonSink())
+      .andThen {
+        case Failure(exception) => logger.info(s"$exception")
+        case Success(value) => logger.info(s"$value: Successufully processing")
+      }
+  }
 
-    val source: Source[Map[String, String], NotUsed] = buildAndValidateSource(inputFile = titleBasics)
+  override def getAllTvSeries(): Unit = {
+
+    val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
     val sink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
 
     val startingTime: Long = System.currentTimeMillis()
@@ -40,13 +91,13 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
   }
 
-  override def getMoviesByTitle(moviePrimaryTitle: String): Unit = {
+  override def getByTvSeriePrimaryTitle(tvSeriePrimaryTitle: String): Unit = {
 
-    val tvSeriesSource: Source[Map[String, String], NotUsed] = buildAndValidateSource(inputFile = titleBasics)
+    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] =
-      filterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
+      filterByTvSeriePrimaryTitleFlow(tvSeriePrimaryTitle = tvSeriePrimaryTitle)
 
     val startTime: Long = System.currentTimeMillis()
     val listTvSeries: Future[Done] = tvSeriesSource
@@ -58,52 +109,30 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
           logger.info(s"$value: successfully processing file, elapsed time: $elapsedTime sec")
         case Failure(error: Error) => logger.error(s"$error")
       }
-  }
 
-  override def getAllPersons(): Unit = {
-    val personSource: Source[Map[String, String], NotUsed] = buildSource(inputFile = nameBasics)
-    val personSink: Sink[Person, Future[Done]] = buildPersonSink(logger = logger)
-
-    //graph
-    val startTime: Long = System.currentTimeMillis()
-    personSource
-      .via(flow = buildPersonFlow())
-      .runWith(sink = personSink)
-      .andThen {
-        case Success(value) =>
-          val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
-          logger.info(s"$value: successfully processing, elapsed time: $elapsedTime sec")
-        case Failure(error: Error) => logger.error(s"$error")
-      }
+    getTvSerieIdByPrimaryTitle(primaryTitle = tvSeriePrimaryTitle)
   }
 
-  override def getPersonById(nconst: String): Unit = {
-    val source: Source[Map[String, String], NotUsed] = buildSource(inputFile = nameBasics)
-
-    val startTime: Long = System.currentTimeMillis()
-    source
-      .via(flow = filterByPersonIdFlow(nconst = nconst))
-      .runWith(sink = buildPersonSink(logger = logger))
-      .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")
-      }
+  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
+      })
+      .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
+    tconst
   }
 
-  override def getPersonByName(primaryName: String): Unit = {
-    val source: Source[Map[String, String], NotUsed] = buildSource(inputFile = nameBasics)
-    source
-      .via(
-        flow = filterByPersonNameFlow(personName = primaryName)
-      )
-      .runWith(sink = buildPersonSink(logger = logger))
-      .andThen {
-        case Failure(exception) => logger.info(s"$exception")
-        case Success(value) => logger.info(s"$value: Successufully processing")
-      }
-  }
+
 }