Merge pull request cfpb#1145 from jmarin/loader-refactor

Nick Grippin · web-flow · commit f48977c3eef8 · 2017-09-29T15:08:34.000-04:00
Loader refactor
diff --git a/api-model/src/main/resources/application.conf b/api-model/src/main/resources/application.conf
@@ -0,0 +1,3 @@
+hmda {
+  connectionFlowParallelism = 5
+}
diff --git a/api-model/src/main/scala/hmda/api/util/FlowUtils.scala b/api-model/src/main/scala/hmda/api/util/FlowUtils.scala
@@ -0,0 +1,52 @@
+package hmda.api.util
+
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.{ HttpMethods, HttpRequest, HttpResponse, Uri }
+import akka.http.scaladsl.unmarshalling.Unmarshal
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{ Flow, Framing }
+import akka.util.ByteString
+import com.typesafe.config.ConfigFactory
+import org.slf4j.Logger
+import scala.concurrent.ExecutionContext
+
+trait FlowUtils {
+
+  implicit val system: ActorSystem
+  implicit val materializer: ActorMaterializer
+  implicit val ec: ExecutionContext
+  val config = ConfigFactory.load()
+  val parallelism = config.getInt("hmda.connectionFlowParallelism")
+
+  def singleConnectionFlow: Flow[HttpRequest, HttpResponse, NotUsed] =
+    Flow[HttpRequest]
+      .mapAsync[HttpResponse](parallelism)(request => {
+        for {
+          response <- Http().singleRequest(request)
+        } yield response
+      })
+
+  def framing: Flow[ByteString, ByteString, NotUsed] = {
+    Framing.delimiter(ByteString("\n"), maximumFrameLength = 65536, allowTruncation = true)
+  }
+
+  def byte2StringFlow: Flow[ByteString, String, NotUsed] =
+    Flow[ByteString].map(bs => bs.utf8String)
+
+  def sendGetRequest(req: String, url: Uri) = {
+    val request = HttpRequest(HttpMethods.GET, uri = s"$url/$req")
+    for {
+      response <- Http().singleRequest(request)
+      content <- Unmarshal(response.entity).to[String]
+    } yield content
+  }
+
+  def exitSys(log: Logger, errorMessage: String, code: Int) = {
+    log.error(errorMessage)
+    system.terminate()
+    Thread.sleep(100)
+    sys.exit(code)
+  }
+}
diff --git a/api/src/main/resources/application-dev.conf b/api/src/main/resources/application-dev.conf
@@ -28,6 +28,15 @@ hmda {
     timeout = ${?HMDA_HTTP_TIMEOUT}
   }
   isDemo = true
+  panel {
+    tcp {
+      host = "0.0.0.0"
+      host = ${?HMDA_PANEL_LOADER_HOST}
+      port = "8888"
+      port = ${?HMDA_PANEL_LOADER_PORT}
+      timeout = 5
+    }
+  }
   actor-flow-parallelism = 4
 }
 
diff --git a/api/src/main/resources/application.conf b/api/src/main/resources/application.conf
@@ -29,6 +29,16 @@ hmda {
   }
   isDemo = false
   isDemo = ${?HMDA_IS_DEMO}
+  panel {
+    tcp {
+      host = "0.0.0.0"
+      host = ${?HMDA_PANEL_LOADER_HOST}
+      port = "8888"
+      port = ${?HMDA_PANEL_LOADER_PORT}
+      timeout = 5
+      parallelism = 5
+    }
+  }
 }
 
 api-dispatcher {
diff --git a/api/src/main/scala/hmda/api/tcp/TcpApi.scala b/api/src/main/scala/hmda/api/tcp/TcpApi.scala
@@ -0,0 +1,39 @@
+package hmda.api.tcp
+
+import java.net.InetSocketAddress
+
+import akka.actor.{ Actor, ActorSystem, Status }
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.Tcp
+import akka.stream.scaladsl.Tcp.ServerBinding
+import hmda.persistence.model.HmdaActor
+
+import scala.concurrent.{ ExecutionContext, Future }
+
+abstract class TcpApi extends HmdaActor {
+
+  val name: String
+  val host: String
+  val port: Int
+
+  implicit val system: ActorSystem
+  implicit val materializer: ActorMaterializer
+  implicit val ec: ExecutionContext
+
+  val tcp: Future[ServerBinding]
+
+  override def receive = {
+    case Tcp.ServerBinding(s) => handleServerBinding(s)
+    case Status.Failure(e) => handleBindFailure(e)
+  }
+
+  private def handleServerBinding(address: InetSocketAddress): Unit = {
+    log.info(s"$name started on {}", address)
+    context.become(Actor.emptyBehavior)
+  }
+
+  private def handleBindFailure(error: Throwable): Unit = {
+    log.error(error, s"Failed to bind to $host:$port")
+    context stop self
+  }
+}
diff --git a/api/src/main/scala/hmda/api/tcp/admin/InstitutionAdminTcpApi.scala b/api/src/main/scala/hmda/api/tcp/admin/InstitutionAdminTcpApi.scala
@@ -0,0 +1,80 @@
+package hmda.api.tcp.admin
+
+import akka.NotUsed
+import akka.pattern.{ ask, pipe }
+import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.stream.ActorMaterializer
+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.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._
+
+object InstitutionAdminTcpApi {
+  def props(supervisor: ActorRef): Props = {
+    Props(new InstitutionAdminTcpApi(supervisor))
+  }
+}
+
+class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils {
+  override val name: String = "hmda-institutions-tcp-api"
+
+  override implicit val system: ActorSystem = context.system
+  override implicit val materializer: ActorMaterializer = ActorMaterializer()
+  override implicit val ec: ExecutionContext = context.dispatcher
+
+  override val host: String = config.getString("hmda.panel.tcp.host")
+  override val port: Int = config.getInt("hmda.panel.tcp.port")
+  val duration = config.getInt("hmda.panel.tcp.timeout").seconds
+  implicit val timeout = Timeout(duration)
+  val buffer = config.getInt("hmda.panel.tcp.parallelism")
+
+  val tcpHandler: Flow[ByteString, ByteString, NotUsed] =
+    Flow[ByteString]
+      .via(framing)
+      .drop(1)
+      .via(byte2StringFlow)
+      .map(x => InstitutionParser(x))
+      .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(
+    tcpHandler,
+    host,
+    port
+  )
+
+  tcp pipeTo self
+
+  private def createInstitution(i: Institution): Future[Institution] = {
+    val fInstitutionsActor = (supervisor ? FindActorByName(InstitutionPersistence.name)).mapTo[ActorRef]
+    for {
+      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
+  }
+
+}
diff --git a/build.sbt b/build.sbt
@@ -126,13 +126,13 @@ lazy val validation = (project in file("validation"))
   ).dependsOn(parserJVM % "compile->compile;test->test")
   .dependsOn(persistenceModel % "compile->compile;test->test")
 
-lazy val panel = (project in file("panel"))
+lazy val loader = (project in file("loader"))
   .settings(hmdaBuildSettings: _*)
   .settings(Revolver.settings:_*)
   .settings(
     Seq(
       assemblyJarName in assembly := {s"${name.value}.jar"},
-      mainClass in assembly := Some("hmda.panel.PanelCsvLoader"),
+      mainClass in assembly := Some("hmda.loader.panel.PanelCsvLoader"),
       assemblyMergeStrategy in assembly := {
         case "application.conf" => MergeStrategy.concat
         case "META-INF/io.netty.versions.properties" => MergeStrategy.concat
diff --git a/cluster/src/main/scala/hmda/cluster/HmdaPlatform.scala b/cluster/src/main/scala/hmda/cluster/HmdaPlatform.scala
@@ -10,6 +10,7 @@ import akka.cluster.http.management.ClusterHttpManagement
 import akka.cluster.singleton.{ ClusterSingletonManager, ClusterSingletonManagerSettings, ClusterSingletonProxy, ClusterSingletonProxySettings }
 import akka.util.Timeout
 import com.typesafe.config.ConfigFactory
+import hmda.api.tcp.admin.InstitutionAdminTcpApi
 import hmda.api.{ HmdaAdminApi, HmdaFilingApi, HmdaPublicApi }
 import hmda.persistence.HmdaSupervisor
 import hmda.persistence.institutions.InstitutionPersistence
@@ -79,6 +80,7 @@ object HmdaPlatform extends App {
     system.actorOf(HmdaFilingApi.props(supervisorProxy, querySupervisorProxy, validationStatsProxy).withDispatcher("api-dispatcher"), "hmda-filing-api")
     system.actorOf(HmdaAdminApi.props(supervisorProxy, querySupervisorProxy).withDispatcher("api-dispatcher"), "hmda-admin-api")
     system.actorOf(HmdaPublicApi.props(querySupervisorProxy).withDispatcher("api-dispatcher"), "hmda-public-api")
+    system.actorOf(InstitutionAdminTcpApi.props(supervisorProxy), "panel-loader-tcp")
   }
 
   //Start Persistence
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -7,6 +7,7 @@ services:
       - '8080:8080'
       - '8081:8081'
       - '8082:8082'
+      - '8888:8888'
     volumes:
       - ./target/scala-2.12/hmda.jar:/opt/hmda.jar
     depends_on:
diff --git a/loader/src/main/resources/application.conf b/loader/src/main/resources/application.conf
@@ -7,5 +7,14 @@ akka {
 hmda {
   adminUrl = "http://0.0.0.0:8081"
   adminUrl = ${?HMDA_HTTP_ADMIN_URL}
+  httpTimeout = 5
+  panel {
+    tcp {
+      host = "127.0.0.1"
+      host = ${?HMDA_PANEL_LOADER_HOST}
+      port = "8888"
+      port = ${?HMDA_PANEL_LOADER_PORT}
+    }
+  }
 }
 
diff --git a/loader/src/main/resources/inst_data_2017.csv b/loader/src/main/resources/inst_data_2017.csv
diff --git a/loader/src/main/resources/inst_data_2017_dummy.csv b/loader/src/main/resources/inst_data_2017_dummy.csv
diff --git a/loader/src/main/resources/logback.xml b/loader/src/main/resources/logback.xml
diff --git a/loader/src/main/scala/hmda/loader/panel/PanelCsvLoader.scala b/loader/src/main/scala/hmda/loader/panel/PanelCsvLoader.scala
@@ -0,0 +1,48 @@
+package hmda.loader.panel
+
+import java.io.File
+
+import akka.actor.ActorSystem
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{ FileIO, Sink, Tcp }
+import org.slf4j.LoggerFactory
+import hmda.api.util.FlowUtils
+
+import scala.concurrent.duration._
+
+object PanelCsvLoader extends FlowUtils {
+  val httpTimeout = config.getInt("hmda.httpTimeout")
+  val adminUrl = config.getString("hmda.adminUrl")
+  val url = s"$adminUrl/institutions"
+  val duration = httpTimeout.seconds
+  override implicit val system: ActorSystem = ActorSystem("hmda-loader")
+  override implicit val materializer = ActorMaterializer()
+  override implicit val ec = system.dispatcher
+  val log = LoggerFactory.getLogger("hmda")
+
+  val host = config.getString("hmda.panel.tcp.host")
+  val port = config.getInt("hmda.panel.tcp.port")
+
+  def main(args: Array[String]): Unit = {
+
+    if (args.length < 1) {
+      exitSys(log, "No file argument provided", 1)
+    }
+
+    val file = new File(args(0))
+    if (!file.exists() || !file.isFile) {
+      exitSys(log, "File does not exist", 2)
+    }
+
+    val connectionFlow = Tcp().outgoingConnection(host, port)
+
+    val source = FileIO.fromPath(file.toPath)
+
+    source
+      .via(connectionFlow)
+      .runWith(Sink.last)
+      .onComplete(_ => system.terminate())
+
+  }
+}
+
diff --git a/panel/src/main/scala/hmda/panel/PanelCsvLoader.scala b/panel/src/main/scala/hmda/panel/PanelCsvLoader.scala

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+hmda {`
	`2`	`+ connectionFlowParallelism = 5`
	`3`	`+}`