last async working but not parsing all file

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

 
 import akka.NotUsed
 import akka.actor.ActorSystem
+import akka.stream.ActorAttributes.supervisionStrategy
+import akka.stream.Supervision.resumingDecider
 import akka.stream.alpakka.csv.scaladsl.{CsvParsing, CsvToMap}
 import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source}
 import akka.util.ByteString
   val titlePrincipalsResource: File = new File("src/main/resources/title.principals.tsv.gz")
   val nameBasicsResource: File = new File("src/main/resources/name.basics.tsv.gz")
 
-  def fileSource(file: File) = {
-    FileIO
-      .fromPath(Paths.get(file.getPath), 1 * 1024 * 1024)
+  def fileSource(file: File): Source[ByteString, NotUsed] = {
+    Source.single(file)
+      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024 * 2))
       .via(Compression.gunzip())
   }
 
   }
 
   //working search not optimized
-  val titleId: String = getIdOfTitle(titleBasicsResource, "Carmencita")
+  val titleId: String = getIdOfTitleAsync(titleBasicsResource, "Carmencita")
   println("value ", titleId)
 
-  val personsIdList: List[String] = getIdOfPersons(titlePrincipalsResource, titleId)
+  val personsIdList: List[String] = getIdOfPersonsAsync(titlePrincipalsResource, titleId).toList.flatten
   println("value ", personsIdList)
 
-  val personsList: List[String] = getPersons(nameBasicsResource, personsIdList)
+  val personsList = getPersons(nameBasicsResource, personsIdList)
   println("value ", personsList)
 
 
-  //working do not change for now
-  def getIdOfTitle(file: File, titleName: String): String = {
-    val result: Future[Option[String]] = Source.single(file)
-      .flatMapConcat(f => FileIO.fromPath(Paths.get(f.getPath), 1 * 1024 * 1024).reduce((a, b) => a ++ b))
-      .via(Compression.gunzip())
-      .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
-      .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
-      .filter(row => row.getOrElse("primaryTitle", "") == titleName)
-      .map(a => a.get("tconst"))
+  def getIdOfTitleAsync(file: File, titleName: String): String = {
+    val result = fileSource(file)
+      .mapAsync(50) {
+        res =>
+          Source.single(res)
+            .via(lineParser)
+            .filter(row => row.getOrElse("primaryTitle", "") == titleName)
+            .map(a => a.get("tconst"))
+            .withAttributes(supervisionStrategy(resumingDecider))
+            .runWith(Sink.head)
+      }
+      .withAttributes(supervisionStrategy(resumingDecider))
       .runWith(Sink.head)
 
     Await.result(result, 5 minutes)
     result.value.get.get.get
   }
 
-  def getIdOfTitleAsync(file: File, titleName: String): immutable.Iterable[Option[String]] = {
-    val result = fileSource(file)
-      .mapAsync(10) {
-        result =>
-          Source.single(result)
+  def getIdOfPersonsAsync(file: File, titleId: String): immutable.Iterable[Option[String]] = {
+    val te = fileSource(file)
+      .mapAsync(50) {
+        res =>
+          Source.single(res)
             .via(lineParser)
-            .filter(row => row.getOrElse("primaryTitle", "") == titleName)
-            .map(a => a.get("tconst"))
+            .filter(row => row.getOrElse("tconst", "") == titleId)
+            .map(a => a.get("nconst"))
+            .withAttributes(supervisionStrategy(resumingDecider))
             .runWith(Sink.collection)
       }
+      .withAttributes(supervisionStrategy(resumingDecider))
       .runWith(Sink.head)
-    Await.result(result, 5 minutes)
+
+    Await.result(te, 5 minutes)
   }
 
-  def getIdOfPersons(file: File, titleId: String): List[String] = {
+  def getPersons(file: File, personsIdList: List[String]): List[String] = {
     val result: Future[immutable.Iterable[Option[String]]] = 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(headerPlaceholder = Option.empty))
-      .filter(row => row.getOrElse("tconst", "") == titleId)
-      .map(a => a.get("nconst"))
+      .filter(row => personsIdList.contains(row.getOrElse("nconst", "")))
+      .map(a => a.get("primaryName"))
       .runWith(Sink.collection)
 
     Await.result(result, 5 minutes)
     result.value.get.get.toList.flatten
   }
 
-  def getPersons(file: File, personsIdList: List[String]): List[String] = {
-    val result: Future[immutable.Iterable[Option[String]]] = Source.single(file)
+ /* def getPersonsAsync(file: File, personsIdList: List[String]) = {
+    val test = 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(headerPlaceholder = Option.empty))
-      .filter(row => personsIdList.contains(row.getOrElse("nconst", "")))
-      .map(a => a.get("primaryName"))
+      .mapAsync(2) {
+        res =>
+          Source.single(res)
+            .via(CsvParsing.lineScanner(CsvParsing.Tab, CsvParsing.DoubleQuote, CsvParsing.DoubleQuote))
+            .via(CsvToMap.toMapAsStringsCombineAll(headerPlaceholder = Option.empty))
+            .filter(row => row.getOrElse("nconst", "") == personsIdList.head)
+            .map(a => a.get("primaryName"))
+            .runWith(Sink.collection)
+      }
       .runWith(Sink.collection)
 
-    Await.result(result, 5 minutes)
-    result.value.get.get.toList.flatten
+    Await.result(test, 5 minutes)
+    test.value.get.get.toList.flatten
   }
-}
+  */
+}