first commit

pom.xml

@@ -65,7 +65,7 @@
             <artifactId>scala-logging-slf4j_2.11</artifactId>
             <version>2.1.2</version>
         </dependency>
-
+        
     </dependencies>
 
     <build>

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

@@ -20,12 +20,12 @@ public class MovieController {
     @RequestMapping(value = "/movies", method = RequestMethod.GET)
     private ResponseEntity<String> getAllMovies() {
         akkaStreamFilesProcessing.getAllMovies();
-        return new ResponseEntity<>("IS RUNNING", HttpStatus.OK);
+        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);
+        return new ResponseEntity<>("is running", HttpStatus.OK);
     }
 }

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

@@ -0,0 +1,9 @@
+package fr.natan.akkastreamfileprocessingapi.models
+
+final case class Person(nconst: String,
+                  primaryName: String,
+                  birthYear: String,
+                  dearthYear: String,
+                  primaryProfession: String,
+                  knownForTitles: List[String]
+                 )

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

@@ -1,6 +1,6 @@
 package fr.natan.akkastreamfileprocessingapi.models
 
-case class TvSeries(
+final case class TvSeries(
                      tconst: String,
                      titleType: String,
                      primaryTitle: String,

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

@@ -19,8 +19,18 @@ object AkkaStreamComponents {
   implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")
 
   private def convertToClassMovie(array: Array[String]): TvSeries = {
-    val movie: TvSeries = TvSeries(array(0), array(1), array(2), array(3), array(4), array(5), array(6), array(7), array(8))
-    movie
+    val tvSerie: TvSeries = TvSeries(
+      array(0),
+      array(1),
+      array(2),
+      array(3),
+      array(4),
+      array(5),
+      array(6),
+      array(7),
+      array(8))
+
+    tvSerie
   }
 
   //flows
@@ -34,10 +44,9 @@ object AkkaStreamComponents {
           convertToClassMovie(movie)
         })
     flow
-
   }
 
-  def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: String): Flow[String, TvSeries, NotUsed] = {
+  def buildfilterByMoviePrimaryTitleFlow(moviePrimaryTitle: String): Flow[String, TvSeries, NotUsed] = {
     val filterFlow: Flow[String, TvSeries, NotUsed] =
       Flow[String]
         .filter((rows: String) => {
@@ -80,14 +89,15 @@ object AkkaStreamComponents {
     if (source == null) {
       throw new FileNotFoundException(filename = inputFile.getPath)
     }
-
     source
   }
 
   //sink
   def buildSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
     val sink = Sink.foreach[TvSeries](
-      (movie: TvSeries) => logger.info(s"${movie.toString}")
+      (movie: TvSeries) => {
+        logger.info(s"${movie.toString}")
+      }
     )
     sink
   }

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

@@ -6,7 +6,7 @@ import akka.{Done, NotUsed}
 import com.typesafe.scalalogging.slf4j.Logger
 import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.titleBasics
 import fr.natan.akkastreamfileprocessingapi.models.TvSeries
-import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildMovieFlow, buildSink, filterByMoviePrimaryTitleFlow}
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildMovieFlow, buildSink, buildfilterByMoviePrimaryTitleFlow}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -20,7 +20,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
   implicit val actorSystem: ActorSystem = ActorSystem("AkkaStreamActor")
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getAllMovies() = {
+  override def getAllMovies(): Unit = {
 
     val source: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
     val sink: Sink[TvSeries, Future[Done]] = buildSink(logger = logger)
@@ -28,6 +28,7 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
     val startingTime: Long = System.currentTimeMillis()
 
+    //graph sink->flow->sink
     source
       .via(flow = buildMovieFlow(movieID = "tconst"))
       .runWith(sink = sink)
@@ -46,17 +47,18 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
     val sink: Sink[TvSeries, Future[Done]] = buildSink(logger = logger)
 
     val filterByMovieTitleFlow: Flow[String, TvSeries, NotUsed] =
-      filterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
+      buildfilterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
 
     val startTime: Long = System.currentTimeMillis()
-    source.via(flow = filterByMovieTitleFlow).runWith(sink = sink).andThen {
+    val listTvSeries: Future[Done]= source.via(flow = filterByMovieTitleFlow)
+      .runWith(sink = sink)
+    .andThen {
       case Success(value) =>
         val elapsedTime: Long = (System.currentTimeMillis() - startTime) / 1000
-        logger.info(s"$value : successfully processing file, elapsed time $elapsedTime sec")
-
-      case Failure(err: Error) => logger.error(s"$err")
+        logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
+      case Failure(error: Error) => logger.error(s"$error")
     }
-
   }
-
 }
+
+