Browse Source

first commit

placidenduwayo 1 year ago
parent
commit
14331e60a4

+ 0
- 12
src/main/scala/fr/natan/akkastreamfileprocessingapi/models/Movie.scala View File

1
-package fr.natan.akkastreamfileprocessingapi.models
2
-
3
-case class Movie( tconst: String, titleType: String, primaryTitle: String,
4
-                originalTitle: String, isAdult: Short, startYear: Int,
5
-                endYear: String, runtimeMinutes: String, genre: String) {
6
-
7
-  override def toString: String = {
8
-    "[movieID : " + tconst +", title-type:"+titleType+", primary-title:"+primaryTitle+"" +
9
-      ", riginal-title:"+originalTitle+", is adult (no=0, yes=1):"+isAdult+", start year:"+startYear+
10
-    ", end year:"+endYear+", runtime minutes:"+runtimeMinutes+", genre:"+genre
11
-  }
12
-}

+ 20
- 0
src/main/scala/fr/natan/akkastreamfileprocessingapi/models/TvSeries.scala View File

1
+package fr.natan.akkastreamfileprocessingapi.models
2
+
3
+case class TvSeries(
4
+                     tconst: String,
5
+                     titleType: String,
6
+                     primaryTitle: String,
7
+                    originalTitle: String,
8
+                     isAdult: String,
9
+                     startYear: String,
10
+                    endYear: String,
11
+                     runtimeMinutes: String,
12
+                     genre: String
13
+                   ) {
14
+
15
+  override def toString: String = {
16
+    "TvSerie[tvSerieID : " + tconst +", title-type:"+titleType+", primary-title:"+primaryTitle+"" +
17
+      ", riginal-title:"+originalTitle+", is adult (no=0, yes=1):"+isAdult+", start year:"+startYear+
18
+    ", end year:"+endYear+", runtime minutes:"+runtimeMinutes+", genre:"+genre+"]"
19
+  }
20
+}

+ 11
- 12
src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamComponents.scala View File

7
 import akka.{Done, NotUsed}
7
 import akka.{Done, NotUsed}
8
 import com.typesafe.scalalogging.slf4j.Logger
8
 import com.typesafe.scalalogging.slf4j.Logger
9
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException
9
 import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException
10
-import fr.natan.akkastreamfileprocessingapi.models.Movie
10
+import fr.natan.akkastreamfileprocessingapi.models.TvSeries
11
 import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists
11
 import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists
12
 
12
 
13
 import java.io.File
13
 import java.io.File
18
 
18
 
19
   implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")
19
   implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")
20
 
20
 
21
-  private def convertToClassMovie(array: Array[String]): Movie = {
22
-    val movie: Movie = Movie(array(0), array(1), array(2), array(3), array(4).toShort, array(5).toInt, array(6), array(7), array(8))
21
+  private def convertToClassMovie(array: Array[String]): TvSeries = {
22
+    val movie: TvSeries = TvSeries(array(0), array(1), array(2), array(3), array(4), array(5), array(6), array(7), array(8))
23
     movie
23
     movie
24
   }
24
   }
25
 
25
 
26
   //flows
26
   //flows
27
 
27
 
28
-  def buildMovieFlow(movieID: String): Flow[String, Movie, NotUsed] = {
28
+  def buildMovieFlow(movieID: String): Flow[String, TvSeries, NotUsed] = {
29
 
29
 
30
-    val flow: Flow[String, Movie, NotUsed] =
30
+    val flow: Flow[String, TvSeries, NotUsed] =
31
       Flow[String].filter(rows => !rows.contains(movieID))
31
       Flow[String].filter(rows => !rows.contains(movieID))
32
         .map(row => {
32
         .map(row => {
33
           val movie: Array[String] = row.split("\t")
33
           val movie: Array[String] = row.split("\t")
37
 
37
 
38
   }
38
   }
39
 
39
 
40
-  def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: String): Flow[String, Movie, NotUsed] = {
41
-    val filterFlow: Flow[String, Movie, NotUsed] =
40
+  def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: String): Flow[String, TvSeries, NotUsed] = {
41
+    val filterFlow: Flow[String, TvSeries, NotUsed] =
42
       Flow[String]
42
       Flow[String]
43
         .filter((rows: String) => {
43
         .filter((rows: String) => {
44
           rows.contains(moviePrimaryTitle)
44
           rows.contains(moviePrimaryTitle)
50
 
50
 
51
     filterFlow
51
     filterFlow
52
   }
52
   }
53
-
54
   //source
53
   //source
55
   def buildSource(inputFile: File): Source[String, NotUsed] = {
54
   def buildSource(inputFile: File): Source[String, NotUsed] = {
56
 
55
 
69
       .via(Compression.gunzip())
68
       .via(Compression.gunzip())
70
       .via(
69
       .via(
71
         Framing.delimiter(ByteString("\n"), 4096)
70
         Framing.delimiter(ByteString("\n"), 4096)
72
-        .map(byteString => byteString.utf8String)
71
+          .map(byteString => byteString.utf8String)
73
       )
72
       )
74
 
73
 
75
     source
74
     source
86
   }
85
   }
87
 
86
 
88
   //sink
87
   //sink
89
-  def buildSink(logger: Logger): Sink[Movie, Future[Done]] = {
90
-    val sink = Sink.foreach[Movie](
91
-      (movie: Movie) => logger.info(s"${movie.toString}")
88
+  def buildSink(logger: Logger): Sink[TvSeries, Future[Done]] = {
89
+    val sink = Sink.foreach[TvSeries](
90
+      (movie: TvSeries) => logger.info(s"${movie.toString}")
92
     )
91
     )
93
     sink
92
     sink
94
   }
93
   }

+ 0
- 2
src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamFileProcessing.scala View File

1
 package fr.natan.akkastreamfileprocessingapi.service
1
 package fr.natan.akkastreamfileprocessingapi.service
2
 
2
 
3
-import fr.natan.akkastreamfileprocessingapi.models.Movie
4
-
5
 trait AkkaStreamFileProcessing {
3
 trait AkkaStreamFileProcessing {
6
 
4
 
7
   def getAllMovies()
5
   def getAllMovies()

+ 9
- 7
src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamFileProcessingImpl.scala View File

5
 import akka.{Done, NotUsed}
5
 import akka.{Done, NotUsed}
6
 import com.typesafe.scalalogging.slf4j.Logger
6
 import com.typesafe.scalalogging.slf4j.Logger
7
 import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.titleBasics
7
 import fr.natan.akkastreamfileprocessingapi.datasource.Datasource.titleBasics
8
-import fr.natan.akkastreamfileprocessingapi.models.Movie
8
+import fr.natan.akkastreamfileprocessingapi.models.TvSeries
9
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildMovieFlow, buildSink, filterByMoviePrimaryTitleFlow}
9
 import fr.natan.akkastreamfileprocessingapi.service.AkkaStreamComponents.{buildAndValidateSource, buildMovieFlow, buildSink, filterByMoviePrimaryTitleFlow}
10
 import org.slf4j.LoggerFactory
10
 import org.slf4j.LoggerFactory
11
 import org.springframework.stereotype.Component
11
 import org.springframework.stereotype.Component
23
   override def getAllMovies() = {
23
   override def getAllMovies() = {
24
 
24
 
25
     val source: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
25
     val source: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
26
-    val sink: Sink[Movie, Future[Done]] = buildSink(logger = logger)
27
-    val movieFlow: Flow[String, Movie, NotUsed] = buildMovieFlow(movieID = "tconst")
26
+    val sink: Sink[TvSeries, Future[Done]] = buildSink(logger = logger)
28
 
27
 
29
-    val startingTime: Long = System.currentTimeMillis()
30
 
28
 
31
-    source.via(flow = movieFlow).runWith(sink = sink)
29
+    val startingTime: Long = System.currentTimeMillis()
32
 
30
 
31
+    source
32
+      .via(flow = buildMovieFlow(movieID = "tconst"))
33
+      .runWith(sink = sink)
33
       .andThen {
34
       .andThen {
34
       case Success(value) =>
35
       case Success(value) =>
35
         val elapsedTime: Long = (System.currentTimeMillis() - startingTime) / 1000
36
         val elapsedTime: Long = (System.currentTimeMillis() - startingTime) / 1000
36
         logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
37
         logger.info(s"$value: successfully processing file, elapsed time $titleBasics: $elapsedTime sec")
37
       case Failure(error: Error) => logger.error(s"$error")
38
       case Failure(error: Error) => logger.error(s"$error")
38
     }
39
     }
40
+
39
   }
41
   }
40
 
42
 
41
   override def getMoviesByTitle(moviePrimaryTitle: String): Unit = {
43
   override def getMoviesByTitle(moviePrimaryTitle: String): Unit = {
42
 
44
 
43
     val source: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
45
     val source: Source[String, NotUsed] = buildAndValidateSource(inputFile = titleBasics)
44
-    val sink: Sink[Movie, Future[Done]] = buildSink(logger = logger)
46
+    val sink: Sink[TvSeries, Future[Done]] = buildSink(logger = logger)
45
 
47
 
46
-    val filterByMovieTitleFlow: Flow[String, Movie, NotUsed] =
48
+    val filterByMovieTitleFlow: Flow[String, TvSeries, NotUsed] =
47
       filterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
49
       filterByMoviePrimaryTitleFlow(moviePrimaryTitle = moviePrimaryTitle)
48
 
50
 
49
     val startTime: Long = System.currentTimeMillis()
51
     val startTime: Long = System.currentTimeMillis()

Powered by TurnKey Linux.