Create filings after creating institutions

Author: jmarin

api/src/main/scala/hmda/api/tcp/admin/InstitutionAdminTcpApi.scala

@@ -8,11 +8,15 @@ import akka.stream.scaladsl.{ Flow, Tcp }
 import akka.util.{ ByteString, Timeout }
 import hmda.api.tcp.TcpApi
 import hmda.api.util.FlowUtils
+import hmda.model.fi.Filing
 import hmda.model.institution.Institution
 import hmda.parser.fi.InstitutionParser
-import hmda.persistence.institutions.InstitutionPersistence
+import hmda.persistence.HmdaSupervisor.FindFilings
+import hmda.persistence.institutions.FilingPersistence.CreateFiling
+import hmda.persistence.institutions.{ FilingPersistence, InstitutionPersistence }
 import hmda.persistence.institutions.InstitutionPersistence.CreateInstitution
 import hmda.persistence.model.HmdaSupervisorActor.FindActorByName
+
 import scala.concurrent.{ ExecutionContext, Future }
 import scala.concurrent.duration._
 
@@ -35,15 +39,14 @@ class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils
   implicit val timeout = Timeout(duration)
   val buffer = config.getInt("hmda.panel.tcp.parallelism")
 
-  val fInstitutionsActor = (supervisor ? FindActorByName(InstitutionPersistence.name)).mapTo[ActorRef]
-
   val tcpHandler: Flow[ByteString, ByteString, NotUsed] =
     Flow[ByteString]
       .via(framing)
       .drop(1)
       .via(byte2StringFlow)
       .map(x => InstitutionParser(x))
-      .mapAsync(parallelism = buffer)(i => createInstitution(fInstitutionsActor, i))
+      .mapAsync(parallelism = buffer)(i => createInstitution(i))
+      .mapAsync(parallelism = buffer)(i => createFiling(i))
       .map(e => ByteString(e.toString))
 
   override val tcp: Future[Tcp.ServerBinding] = Tcp().bindAndHandle(
@@ -54,11 +57,24 @@ class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils
 
   tcp pipeTo self
 
-  private def createInstitution(fActor: Future[ActorRef], i: Institution): Future[Institution] = {
+  private def createInstitution(i: Institution): Future[Institution] = {
+    val fInstitutionsActor = (supervisor ? FindActorByName(InstitutionPersistence.name)).mapTo[ActorRef]
     for {
-      actor <- fActor
-      i <- (actor ? CreateInstitution(i)).mapTo[Option[Institution]].map(i => i.getOrElse(Institution.empty))
+      actor <- fInstitutionsActor
+      i <- (actor ? CreateInstitution(i))
+        .mapTo[Option[Institution]]
+        .map(i => i.getOrElse(Institution.empty))
     } yield i
   }
 
+  private def createFiling(institution: Institution): Future[Filing] = {
+    val fFilingPersistence = (supervisor ? FindFilings(FilingPersistence.name, institution.id)).mapTo[ActorRef]
+    for {
+      actor <- fFilingPersistence
+      f <- (actor ? CreateFiling(Filing(institution.activityYear.toString, institution.id)))
+        .mapTo[Option[Filing]]
+        .map(x => x.getOrElse(Filing()))
+    } yield f
+  }
+
 }