testing

pom.xml

@@ -88,6 +88,11 @@
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-stream_${scala.binary.version}</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.lightbend.akka</groupId>
+            <artifactId>akka-stream-alpakka-csv_${scala.binary.version}</artifactId>
+            <version>4.0.0</version>
+        </dependency>
 
         <!-- alpakka -->
         <dependency>

src/main/resources/test.tsv

@@ -0,0 +1,17 @@
+tconst	titleType	primaryTitle	originalTitle	isAdult	startYear	endYear	runtimeMinutes	genres
+tt0000001	short	Carmencita	Carmencita	0	1894	\N	1	Documentary,Short
+tt0000002	short	Le clown et ses chiens	Le clown et ses chiens	0	1892	\N	5	Animation,Short
+tt0000003	short	Pauvre Pierrot	Pauvre Pierrot	0	1892	\N	4	Animation,Comedy,Romance
+tt0000004	short	Un bon bock	Un bon bock	0	1892	\N	12	Animation,Short
+tt0000005	short	Blacksmith Scene	Blacksmith Scene	0	1893	\N	1	Comedy,Short
+tt0000006	short	Chinese Opium Den	Chinese Opium Den	0	1894	\N	1	Short
+tt0000007	short	Corbett and Courtney Before the Kinetograph	Corbett and Courtney Before the Kinetograph	0	1894	\N	1	Short,Sport
+tt0000008	short	Edison Kinetoscopic Record of a Sneeze	Edison Kinetoscopic Record of a Sneeze	0	1894	\N	1	Documentary,Short
+tt0000009	movie	Miss Jerry	Miss Jerry	0	1894	\N	45	Romance
+tt0000010	short	Leaving the Factory	La sortie de l'usine Lumi?¿re ?á Lyon	0	1895	\N	1	Documentary,Short
+tt0000011	short	Akrobatisches Potpourri	Akrobatisches Potpourri	0	1895	\N	1	Documentary,Short
+tt0000012	short	The Arrival of a Train	L'arriv?®e d'un train ?á La Ciotat	0	1896	\N	1	Documentary,Short
+tt0000013	short	The Photographical Congress Arrives in Lyon	Le d?®barquement du congr?¿s de photographie ?á Lyon	0	1895	\N	1	Documentary,Short
+tt0000014	short	The Waterer Watered	L'arroseur arros?®	0	1895	\N	1	Comedy,Short
+tt0000015	short	Autour d'une cabine	Autour d'une cabine	0	1894	\N	2	Animation,Short
+tt0000016	short	Boat Leaving the Port	Barque sortant du port	0	1895	\N	1	Documentary,Short

src/main/resources/test2.tsv

@@ -0,0 +1,2 @@
+tconst	titleType	name	originalTitle	isAdult	startYear	endYear	runtimeMinutes	ge
+tt0000015	short	Antoine	Autour d'une cabine	0	1900	\N	2	Animation,Short

src/main/scala/com/mediahub/FileReader.scala

@@ -0,0 +1,79 @@
+package com.mediahub
+
+import akka.{Done, NotUsed}
+import org.springframework.stereotype.Component
+
+import java.io.File
+import akka.actor.ActorSystem
+import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap}
+import akka.stream.javadsl.FileIO
+import akka.stream.scaladsl.{Compression, Flow, Framing, Keep, Sink, Source}
+import akka.util.ByteString
+
+import java.nio.file.Paths
+import java.util.stream.Collectors
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{ExecutionContext, Future}
+
+
+@Component
+class FileReader {
+
+  implicit val system: ActorSystem = ActorSystem()
+
+  val list: List[File] = getListOfFiles
+  val source: Source[File, NotUsed] = Source(list)
+
+  val result = source
+    .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+    .via(CsvParsing.lineScanner(CsvParsing.Tab))
+    .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
+    .filter(row => row.getOrElse("primaryTitle", "") == "Autour d'une cabine")
+    .map(a => a.get("tconst"))
+    .toMat(Sink.foreach(println))(Keep.both)
+   // .runForeach(h => println("source 1 ", h))
+
+  //.runWith(Sink.seq)
+
+  source
+    .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+    .via(CsvParsing.lineScanner(CsvParsing.Tab))
+    .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
+    .runForeach(h => println("source 2 ", h))
+   /* .filter(row => row.getOrElse("tconst", "name") == (result.value, "Antoine"))
+    .runForeach(println)*/
+
+
+
+
+  // readAsync()
+
+
+  /* def readAsync(): Source = {
+     Source(list)
+       .mapAsync(4) { value =>
+         implicit val ec: ExecutionContext = ActorSystem().dispatcher
+         Future {
+           source
+             .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+             .via(CsvParsing.lineScanner(CsvParsing.Tab))
+             .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
+             .filter(row => row.getOrElse("primaryTitle", "") == "Autour d'une cabine")
+             .runForeach(println)
+           println(s"Process in Thread:${Thread.currentThread().getName}")
+           value
+         }
+       }
+       .runWith(Sink.foreach(value => println(s"Item emitted:$value in Thread:${Thread.currentThread().getName}")))
+   }*/
+
+  def getListOfFiles: List[File] = {
+    val d = new File("src/main/resources")
+    if (d.exists && d.isDirectory) {
+      d.listFiles.filter(_.isFile).toList
+    } else {
+      List[File]()
+    }
+  }
+
+}

src/main/scala/com/mediahub/HelloWorld.scala

@@ -1,17 +0,0 @@
-package com.mediahub
-
-import org.springframework.stereotype.Component
-
-@Component
-class HelloWorld {
-  println("Hello from default constructor")
-  greet()
-
-
-
-  def greet() = {
-    val greeting = "Hello from SCALA!!!!"
-    println(greeting)
-  }
-
-}

src/main/scala/com/mediahub/MovieQueryService.scala

@@ -0,0 +1,27 @@
+package com.mediahub
+
+import akka.stream.scaladsl.Source
+
+class MovieQueryService {
+
+  final case class Principal(
+                              name: String,
+                              birthYear: Int,
+                              deathYear: Option[Int],
+                              profession: List[String])
+
+  final case class TvSeries(
+                             original: String,
+                             startYear: Int,
+                             endYear: Option[Int],
+                             genres: List[String])
+
+  trait MovieService {
+    def principalsForMovieName(name: String): Source[Principal, _]
+
+    def tvSeriesWithGreatestNumberOfEpisodes(): Source[TvSeries, _]
+  }
+
+}
+
+