first commit

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

@@ -35,11 +35,17 @@ public class MovieController {
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 
-    @RequestMapping(value = "/persons/{personID}", method = RequestMethod.GET)
+    @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);
 
     }
+
+    @RequestMapping(value = "/persons/name/{primaryName}", method = RequestMethod.GET)
+    private ResponseEntity<String> getPersonByName(@PathVariable(name = "primaryName") String primaryName){
+        akkaStreamFilesProcessing.getPersonByName(primaryName);
+        return new ResponseEntity<>("is running", HttpStatus.OK);
+    }
 }

src/main/scala/fr/natan/akkastreamfileprocessingapi/jsonformat/ToJsonMapping.scala

@@ -1,5 +0,0 @@
-package fr.natan.akkastreamfileprocessingapi.jsonformat
-
-case class ToJsonMapping() {
-
-}

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

@@ -0,0 +1,48 @@
+package fr.natan.akkastreamfileprocessingapi.models
+
+object Models {
+
+  final case class Person(
+                           nconst: String,
+                           primaryName: String,
+                           birthYear: String,
+                           deathYear: String,
+                           primaryProfession: List[String],
+                           knownForTitles: List[String]
+                         ) {
+    override def toString: String = {
+      "Person[person-ID:" + nconst +
+        ", primary-name:" + primaryName +
+        ", birth-year:" + birthYear +
+        ", dearth year:" + deathYear +
+        ", primary profession:" + primaryProfession +
+        ", known for titles:" + knownForTitles +
+        "]"
+    }
+  }
+
+  final case class TvSeries(
+                             tconst: String,
+                             titleType: String,
+                             primaryTitle: String,
+                             originalTitle: String,
+                             isAdult: String,
+                             startYear: String,
+                             endYear: String,
+                             runtimeMinutes: String,
+                             genres: String
+                           ) {
+
+    override def toString: String = {
+      "TvSerie[tvSerieID : " + tconst +
+        ", 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 + "]"
+    }
+  }
+}

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

@@ -0,0 +1,30 @@
+package fr.natan.akkastreamfileprocessingapi.models
+import Models.{Person, TvSeries}
+object ModelsBuilder {
+
+  def buildPersonModel(map: Map[String, String]): Person = {
+    Person(
+      map("nconst"),
+      map("primaryName"),
+      map("birthYear"),
+      map("deathYear"),
+      map("primaryProfession").split(",").toList,
+      map("knownForTitles").split(",").toList,
+    )
+  }
+
+  def buildTvSerieModel(map: 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")
+    )
+    tvSerie
+  }
+}

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

@@ -1,20 +0,0 @@
-package fr.natan.akkastreamfileprocessingapi.models
-
-final case class Person(
-                         nconst: String,
-                         primaryName: String,
-                         birthYear: String,
-                         deathYear: String,
-                         primaryProfession: List[String],
-                         knownForTitles: List[String]
-                       ){
-  override def toString: String = {
-    "Person[ person-ID:"+nconst+
-      ", primary-name:"+primaryName+
-    ", birth-year:"+birthYear+
-    ", dearth year:"+deathYear+
-    ", primary profession:"+primaryProfession+
-    ", known for titles:"+knownForTitles+
-    "]"
-  }
-}

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

@@ -1,26 +0,0 @@
-package fr.natan.akkastreamfileprocessingapi.models
-
-final case class TvSeries(
-                     tconst: String,
-                     titleType: String,
-                     primaryTitle: String,
-                    originalTitle: String,
-                     isAdult: String,
-                     startYear: String,
-                    endYear: String,
-                     runtimeMinutes: String,
-                     genres: String
-                   ) {
-
-  override def toString: String = {
-    "TvSerie[tvSerieID : " + tconst +
-      ", 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+"]"
-  }
-}

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

@@ -2,15 +2,13 @@ package fr.natan.akkastreamfileprocessingapi.service
 
 import akka.actor.ActorSystem
 import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap}
-import akka.stream.javadsl.Framing
 import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source}
-import akka.util.ByteString
 import akka.{Done, NotUsed}
 import com.typesafe.scalalogging.slf4j.Logger
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException
-import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.separator
-import fr.natan.akkastreamfileprocessingapi.models.{Person, TvSeries}
+import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
 import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists
+import fr.natan.akkastreamfileprocessingapi.models.ModelsBuilder.{buildPersonModel, buildTvSerieModel}
 
 import java.io.File
 import java.nio.file.Paths
@@ -20,81 +18,71 @@ object AkkaStreamComponents {
 
   implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")
 
-  private def convertToTvSerie(map: 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")
-    )
-
-    tvSerie
-  }
-
-  private def convertToPerson(map: Map[String, String]): Person ={
-    Person(
-      map("nconst"),
-      map("primaryName"),
-      map("birthYear"),
-      map("deathYear"),
-      map("primaryProfession").split(",").toList,
-      map("knownForTitles").split(",").toList,
-    )
-  }
-
-  //flows
+  //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 => {
-          convertToTvSerie(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]]
         .map((rowMap: Map[String, String]) => {
-          convertToPerson(rowMap)
+          buildPersonModel(rowMap)
         })
 
     personFlow
   }
 
+  //flow3
   def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: 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
         })
         .map(rowMap => {
-          convertToTvSerie(map = rowMap)
+          buildTvSerieModel(map = rowMap)
         })
 
     filterFlow
   }
 
-  def filterByPersonID(nconst: String): Flow[Map[String, String], Person, NotUsed]={
+  //flow4
+  def filterByPersonIdFlow(nconst: 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("nconst","")==nconst
         })
         .map(rowMap=>{
-          convertToPerson(map = rowMap)
+          buildPersonModel(map = rowMap)
+        })
+
+    personFilter
+  }
+
+  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
+        })
+        .map((rowMap: Map[String, String])=>{
+          buildPersonModel(map = rowMap)
         })
 
     personFilter
   }
 
-  //source
+  //source building
   def buildSource(inputFile: File): Source[Map[String, String], NotUsed] = {
 
     var datasource: Source[Map[String, String], NotUsed] = null
@@ -126,7 +114,9 @@ object AkkaStreamComponents {
     source
   }
 
-  //sink
+  //sinks building
+
+  //sink1
   def buildTvSeriesSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
     val tvSeriesSink : Sink[TvSeries, Future[Done]]= Sink.foreach[TvSeries](
       (movie: TvSeries) => {
@@ -136,15 +126,14 @@ object AkkaStreamComponents {
     tvSeriesSink
   }
 
-  def buildPersonsSink(logger: Logger): Sink[Person,Future[Done]] = {
-    val listPersonsSink: Sink[Person, Future[Done]]=
-      Sink.foreach[Person]((person: Person)=>{
-        logger.info(s"${person.toString}")
-      })
-
+  //sink2
+  def buildAllPersonsSink(logger: Logger): Sink[Person,Future[Seq[Person]]] = {
+    val listPersonsSink: Sink[Person, Future[Seq[Person]]]=
+      Sink.seq
     listPersonsSink
   }
 
+  //sink3
   def buildPersonSink(logger: Logger): Sink[Person, Future[Done]] = {
     Sink.foreach[Person](
       (person: Person) => logger.info(s"${person.toString}")

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

@@ -1,9 +1,5 @@
 package fr.natan.akkastreamfileprocessingapi.service
 
-import akka.Done
-
-import scala.concurrent.Future
-
 trait AkkaStreamFileProcessing {
 
   def getAllMovies()
@@ -12,4 +8,5 @@ trait AkkaStreamFileProcessing {
   def getAllPersons()
 
   def getPersonById(nconst: String)
+  def getPersonByName(primaryName: String)
 }

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

@@ -5,16 +5,8 @@ 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.models.{Person, TvSeries}
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{
-  buildAndValidateSource,
-  buildPersonFlow,
-  buildPersonSink,
-  buildPersonsSink,
-  buildSource, buildTvSerieFlow,
-  buildTvSeriesSink, filterByMoviePrimaryTitleFlow,
-  filterByPersonID
-}
+import fr.natan.akkastreamfileprocessingapi.models.Models.{Person, TvSeries}
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildPersonFlow, buildPersonSink, buildSource, buildTvSerieFlow, buildTvSeriesSink, filterByMoviePrimaryTitleFlow, filterByPersonIdFlow, filterByPersonNameFlow}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -33,7 +25,6 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     val source: Source[Map[String, String], NotUsed] = buildAndValidateSource(inputFile = titleBasics)
     val sink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
 
-
     val startingTime: Long = System.currentTimeMillis()
 
     //graph sink->flow->sink
@@ -51,10 +42,10 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
   override def getMoviesByTitle(moviePrimaryTitle: String): Unit = {
 
-    /*val tvSeriesSource: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
+    val tvSeriesSource: Source[Map[String, String], NotUsed] = buildAndValidateSource(inputFile = titleBasics)
     val tvSeriesSink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
 
-    val filterByMovieTitleFlow: Flow[String, TvSeries, NotUsed] =
+    val filterByMovieTitleFlow: Flow[Map[String, String], TvSeries, NotUsed] =
       filterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
 
     val startTime: Long = System.currentTimeMillis()
@@ -64,14 +55,14 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
       .andThen {
         case Success(value) =>
           val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
-          logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
+          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]] = buildPersonsSink(logger = logger)
+    val personSink: Sink[Person, Future[Done]] = buildPersonSink(logger = logger)
 
     //graph
     val startTime: Long = System.currentTimeMillis()
@@ -81,29 +72,38 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
       .andThen {
         case Success(value) =>
           val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
-          logger.info(s"$value: successfully processing file $elapsedTime sec")
+          logger.info(s"$value: successfully processing, elapsed time: $elapsedTime sec")
         case Failure(error: Error) => logger.error(s"$error")
       }
   }
 
   override def getPersonById(nconst: String): Unit = {
     val source: Source[Map[String, String], NotUsed] = buildSource(inputFile = nameBasics)
-    val sink: Sink[Person, Future[Done]] = buildPersonSink(logger = logger)
 
     val startTime: Long = System.currentTimeMillis()
     source
-      .via(flow = filterByPersonID(nconst = nconst))
-      .runWith(sink = sink)
+      .via(flow = filterByPersonIdFlow(nconst = nconst))
+      .runWith(sink = buildPersonSink(logger = logger))
       .andThen {
-        case Success(value) => {
+        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 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")
+      }
+  }
 }