2nd method not tested

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

@@ -12,7 +12,7 @@ import org.springframework.stereotype.Component
 import java.io.File
 import java.nio.charset.StandardCharsets
 import java.nio.file.Paths
-import scala.collection.immutable
+import scala.collection.{immutable, mutable}
 import scala.concurrent.Await
 import scala.concurrent.duration.DurationInt
 import scala.language.postfixOps
@@ -26,6 +26,7 @@ class MovieQueryService extends MovieServiceTrait {
   val titleBasicsResource: File = new File("src/main/resources/title.basics.tsv.gz")
   val titlePrincipalsResource: File = new File("src/main/resources/title.principals.tsv.gz")
   val nameBasicsResource: File = new File("src/main/resources/name.basics.tsv.gz")
+  val titleEpisodeResource: File = new File("src/main/resources/title.episode.tsv.gz")
 
   def fileSource(file: File): Source[ByteString, NotUsed] = {
     Source.single(file)
@@ -40,6 +41,10 @@ class MovieQueryService extends MovieServiceTrait {
   }
 
 
+  val res = getSeriesTitle(titleEpisodeResource)
+  val r = getTitleByIds(titleBasicsResource, res)
+  println(r)
+
   override def principalsForMovieName(name: String): List[Principal] = {
 
     val titleId: String = getIdOfTitle(titleBasicsResource, name)
@@ -49,6 +54,8 @@ class MovieQueryService extends MovieServiceTrait {
   }
 
   override def tvSeriesWithGreatestNumberOfEpisodes(): List[TvSeries] = ???
+    //val personsIdList: List[String] = getSeriestitle(titleEpisodeResource)
+
 
 
   def getIdOfTitle(file: File, titleName: String): String = {
@@ -69,6 +76,23 @@ class MovieQueryService extends MovieServiceTrait {
     result.value.get.get.get
   }
 
+  def getTitleByIds(file: File, titleIdMap: mutable.Map[String, Int]): Seq[Option[String]] = {
+    val result = fileSource(file)
+      .mapAsync(50) {
+        res =>
+          Source.single(res)
+            .via(lineParser)
+            .filter(row => titleIdMap.contains(row.getOrElse("tconst", "")))
+            .map(a => a.get("primaryTitle"))
+            .withAttributes(supervisionStrategy(resumingDecider))
+            .runWith(Sink.collection)
+      }
+      .withAttributes(supervisionStrategy(resumingDecider))
+      .runWith(Sink.head)
+
+    Await.result(result, 5 minutes)
+    result.value.get.get.toList
+  }
   def getIdOfPersons(file: File, titleId: String): immutable.Iterable[Option[String]] = {
     val te = fileSource(file)
       .mapAsync(50) {
@@ -101,4 +125,26 @@ class MovieQueryService extends MovieServiceTrait {
     result.value.get.get.toList
   }
 
+  def getSeriesTitle(file: File): mutable.Map[String, Int] = {
+
+    val seriesMap = collection.mutable.Map("0" -> 0)
+    val maxSeriesMap = collection.mutable.Map("0" -> 0)
+
+    val result = Source.single(file)
+      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024))
+      .via(Compression.gunzip())
+      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
+      .via(CsvToMap.toMapAsStringsCombineAll(StandardCharsets.UTF_8, Option.empty))
+      .map(row => seriesMap += seriesMap.get(row("parentTconst")).map(x => row("parentTconst") -> (x + 1)).getOrElse(row("parentTconst") -> 1))
+      .runWith(Sink.seq)
+
+    Await.result(result, 5 minutes)
+
+    for (_ <- 1 to 10) {
+      maxSeriesMap.addOne(seriesMap.maxBy(x => x._2)._1 -> seriesMap.maxBy(x => x._2)._2)
+      seriesMap.remove(seriesMap.maxBy(x => x._2)._1)
+    }
+    maxSeriesMap
+  }
+
 }