Merge pull request cfpb#1131 from schbetsy/parsing-messages

Fix SubmissionSummary Intermittent Spec Failure

Author: jmarin

model/shared/src/test/scala/hmda/model/institution/SubmissionGenerators.scala

@@ -34,7 +34,8 @@ object SubmissionGenerators {
       status <- submissionStatusGen
       start <- Gen.choose(1483287071000L, 1514736671000L)
       end <- Gen.choose(1483287071000L, 1514736671000L)
+      fileName <- Gen.alphaStr
       receipt <- Gen.alphaStr
-    } yield Submission(id, status, start, end, receipt)
+    } yield Submission(id, status, start, end, fileName, receipt)
   }
 }

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))
@@ -62,11 +64,10 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
   var state = HmdaFileParseState()
   var encounteredParsingErrors: Boolean = false
   val manager = context.parent
-  val statRef = for {
-    stat <- (manager ? GetActorRef(SubmissionLarStats.name)).mapTo[ActorRef]
-  } yield {
-    stat
-  }
+  val statRef = (manager ? GetActorRef(SubmissionLarStats.name)).mapTo[ActorRef]
+
+  var tsParsingDone: Boolean = false
+  var larParsingDone: Boolean = false
 
   override def updateState(event: Event): Unit = {
     state = state.updated(event)
@@ -77,7 +78,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 +91,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] }
@@ -100,7 +100,7 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
         .zip(Source.fromIterator(() => Iterator.from(2)))
         .map {
           case (lar, index) =>
-            sendLar(lar)
+            statRef.map(_ ! lar)
             LarCsvParser(lar, index)
         }
         .map {
@@ -112,7 +112,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 +140,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
@@ -164,11 +169,5 @@ class HmdaFileParser(submissionId: SubmissionId) extends HmdaPersistentActor {
       context stop self
 
   }
-
-  private def sendLar(s: String) {
-    for {
-      stat <- statRef
-    } yield stat ! s
-  }
 }
 

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

@@ -139,45 +139,42 @@ class HmdaFileValidator(supervisor: ActorRef, validationStats: ActorRef, submiss
   override def receiveCommand: Receive = {
 
     case BeginValidation(replyTo) =>
-      val validationStarted = ValidationStarted(submissionId)
-      sender() ! validationStarted
+      sender() ! ValidationStarted(submissionId)
       events(parserPersistenceId)
         .filter(x => x.isInstanceOf[TsParsed])
-        .map(e => e.asInstanceOf[TsParsed].ts)
-        .map(ts => (ts, validateTs(ts, ctx).toEither))
+        .map { e => e.asInstanceOf[TsParsed].ts }
+        .map { ts =>
+          self ! ts
+          validationStats ! AddSubmissionTaxId(ts.taxId, submissionId)
+          self ! ValidateAggregate(ts)
+          validateTs(ts, ctx).toEither
+        }
         .map {
-          case (_, Right(ts)) =>
-            validationStats ! AddSubmissionTaxId(ts.taxId, submissionId)
-            ValidateAggregate(ts)
-          case (ts, Left(errors)) =>
-            validationStats ! AddSubmissionTaxId(ts.taxId, submissionId)
-            self ! ValidateAggregate(ts)
-            TsValidationErrors(errors.list.toList)
+          case Right(_) => // do nothing
+          case Left(errors) => TsValidationErrors(errors.list.toList)
         }
         .runWith(Sink.actorRef(self, NotUsed))
 
       val larSource: Source[LoanApplicationRegister, NotUsed] = events(parserPersistenceId)
         .filter(x => x.isInstanceOf[LarParsed])
         .map(e => e.asInstanceOf[LarParsed].lar)
 
-      larSource.map(lar => (lar, validateLar(lar, ctx).toEither))
+      larSource.map { lar =>
+        self ! lar
+        validateLar(lar, ctx).toEither
+      }
         .map {
-          case (_, Right(l)) => l
-          case (lar, Left(errors)) => {
-            self ! lar
-            LarValidationErrors(errors.list.toList)
-          }
+          case Right(_) => // do nothing
+          case Left(errors) => LarValidationErrors(errors.list.toList)
         }
         .runWith(Sink.actorRef(self, ValidateMacro(larSource, replyTo)))
 
     case ValidateAggregate(ts) =>
       performAsyncChecks(ts, ctx)
         .map(validations => validations.toEither)
         .map {
-          case Right(_) => self ! ts
-          case Left(errors) =>
-            self ! TsValidationErrors(errors.list.toList)
-            self ! ts
+          case Right(_) => // do nothing
+          case Left(errors) => self ! TsValidationErrors(errors.list.toList)
         }
 
     case ts: TransmittalSheet =>