first commit

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

@@ -28,4 +28,10 @@ public class MovieController {
         akkaStreamFilesProcessing.getMoviesByTitle(movieTitle);
         return new ResponseEntity<>("is running", HttpStatus.OK);
     }
+
+    @RequestMapping(value = "/persons", method = RequestMethod.GET)
+    private ResponseEntity<String> getallPersons(){
+        akkaStreamFilesProcessing.getAllPersons();
+        return new ResponseEntity<>("is running", HttpStatus.OK);
+    }
 }

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

@@ -4,9 +4,9 @@ import akka.actor.ActorSystem
 import akka.stream.scaladsl.{Flow, Sink, Source}
 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, buildTvSerieFlow, buildTvSeriesSink, buildfilterByMoviePrimaryTitleFlow}
+import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.{nameBasics, titleBasics}
+import fr.natan.akkastreamfileprocessingapi.models.{Person, TvSeries}
+import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildPersonFlow, buildPersonSink, buildSource, buildTvSerieFlow, buildTvSeriesSink, buildfilterByMoviePrimaryTitleFlow}
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 
@@ -43,16 +43,17 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
 
   override def getMoviesByTitle(moviePrimaryTitle: String): Unit = {
 
-    val source: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
-    val sink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
+    val tvSeriesSource: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
+    val tvSeriesSink: Sink[TvSeries, Future[Done]] = buildTvSeriesSink(logger = logger)
 
     val filterByMovieTitleFlow: Flow[String, TvSeries, NotUsed] =
       buildfilterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
 
     val startTime: Long = System.currentTimeMillis()
-    val listTvSeries: Future[Done]= source.via(flow = filterByMovieTitleFlow)
-      .runWith(sink = sink)
-    .andThen {
+    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 $titleBasics: $elapsedTime sec")
@@ -61,7 +62,18 @@ class AkkaStreamFileProcessingImpl extends AkkaStreamFileProcessing {
   }
 
   override def getAllPersons(): Unit = {
-    ???
+    val personSource: Source[String, NotUsed]= buildSource(inputFile = nameBasics)
+    val personSink: Sink[Person, Future[Done]] = buildPersonSink(logger = logger)
+
+    //graph
+    personSource
+      .via(flow = buildPersonFlow(personID = "nconst"))
+      .runWith(sink = personSink)
+      .andThen {
+        case Success(value) =>
+          logger.info(s"$value: successfully processing file")
+        case Failure(error: Error) => logger.error(s"$error")
+      }
   }
 }