Ensure all parsing is finished before beginning validation

Author: schbetsy

persistence/src/main/scala/hmda/persistence/processing/HmdaFileParser.scala

@@ -28,6 +28,8 @@ object HmdaFileParser {
 
   case class ReadHmdaRawFile(persistenceId: String, replyTo: ActorRef) extends Command
   case class FinishParsing(replyTo: ActorRef) extends Command
+  case class FinishParsingTS(replyTo: ActorRef) extends Command
+  case class FinishParsingLARs(replyTo: ActorRef) extends Command
   case class GetStatePaginated(page: Int)
 
   def props(id: SubmissionId): Props = Props(new HmdaFileParser(id))
@@ -68,6 +70,9 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
     stat
   }
 
+  var tsParsingDone: Boolean = false
+  var larParsingDone: Boolean = false
+
   override def updateState(event: Event): Unit = {
     state = state.updated(event)
   }
@@ -77,7 +82,6 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
   override def receiveCommand: Receive = {
 
     case ReadHmdaRawFile(persistenceId, replyTo: ActorRef) =>
-
       val parsedTs = events(persistenceId)
         .filter { x => x.isInstanceOf[LineAdded] }
         .map { case LineAdded(_, data) => data }
@@ -91,7 +95,7 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
         }
 
       parsedTs
-        .runForeach(pTs => self ! pTs)
+        .runWith(Sink.actorRef(self, FinishParsingTS(replyTo)))
 
       val parsedLar = events(persistenceId)
         .filter { x => x.isInstanceOf[LineAdded] }
@@ -112,7 +116,7 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
 
       parsedLar
         .mapAsync(parallelism = flowParallelism)(x => (self ? x).mapTo[Persisted.type])
-        .runWith(Sink.actorRef(self, FinishParsing(replyTo)))
+        .runWith(Sink.actorRef(self, FinishParsingLARs(replyTo)))
 
     case tp @ TsParsed(ts) =>
       persist(tp) { e =>
@@ -140,11 +144,16 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
         sender() ! Persisted
       }
 
-    case FinishParsing(replyTo) =>
-      for {
-        stat <- statRef
-      } yield stat ! CountSubmittedLarsInSubmission
+    case FinishParsingTS(replyTo) =>
+      tsParsingDone = true
+      if (larParsingDone) self ! FinishParsing(replyTo)
 
+    case FinishParsingLARs(replyTo) =>
+      larParsingDone = true
+      if (tsParsingDone) self ! FinishParsing(replyTo)
+
+    case FinishParsing(replyTo) =>
+      statRef.map(_ ! CountSubmittedLarsInSubmission)
       if (encounteredParsingErrors)
         replyTo ! ParsingCompletedWithErrors(submissionId)
       else