Refactor panel -> loader project

Author: jmarin

api-model/src/main/resources/application.conf

@@ -0,0 +1,3 @@
+hmda {
+  connectionFlowParallelism = 5
+}

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)
+  }
+}

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

panel/src/main/resources/application.conf renamed to loader/src/main/resources/application.conf

@@ -7,5 +7,6 @@ akka {
 hmda {
   adminUrl = "http://0.0.0.0:8081"
   adminUrl = ${?HMDA_HTTP_ADMIN_URL}
+  httpTimeout = 5
 }
 

panel/src/main/resources/inst_data_2017.csv renamed to loader/src/main/resources/inst_data_2017.csv

@@ -0,0 +1,23 @@
+package hmda.loader.http
+
+import akka.NotUsed
+import akka.http.scaladsl.model._
+import akka.stream.scaladsl.Flow
+import akka.util.ByteString
+import hmda.api.protocol.admin.WriteInstitutionProtocol
+import hmda.parser.fi.InstitutionParser
+import spray.json._
+
+trait InstitutionFlowUtils extends WriteInstitutionProtocol {
+
+  def institutionStringToHttpFlow(url: Uri): Flow[String, HttpRequest, NotUsed] =
+    Flow[String]
+      .map { x =>
+        val payload = ByteString(InstitutionParser(x).toJson.toString)
+        HttpRequest(
+          HttpMethods.POST,
+          uri = url,
+          entity = HttpEntity(MediaTypes.`application/json`, payload)
+        )
+      }
+}

panel/src/main/resources/inst_data_2017_dummy.csv renamed to loader/src/main/resources/inst_data_2017_dummy.csv

@@ -0,0 +1,66 @@
+package hmda.panel
+
+import java.io.File
+
+import akka.actor.ActorSystem
+import akka.http.scaladsl.model.HttpResponse
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{ FileIO, Framing, Sink }
+import akka.util.ByteString
+import hmda.loader.http.InstitutionFlowUtils
+import org.slf4j.LoggerFactory
+import hmda.api.util.FlowUtils
+import scala.concurrent.duration._
+
+object PanelCsvLoader extends FlowUtils with InstitutionFlowUtils {
+  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")
+
+  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)
+    }
+
+    sendGetRequest("delete", url).map(_ => {
+      val response = sendGetRequest("create", url)
+
+      val source = FileIO.fromPath(file.toPath)
+
+      response.onComplete(s =>
+        if (s.getOrElse("").equals("InstitutionSchemaCreated()")) {
+          log.info(s.get)
+          val completedF = source
+            .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 1024, allowTruncation = true))
+            .drop(1)
+            .via(byte2StringFlow)
+            .via(institutionStringToHttpFlow(url))
+            .via(singleConnectionFlow)
+            .runWith(Sink.foreach[HttpResponse](elem => log.info(elem.entity.toString)))
+
+          completedF.onComplete(result => {
+            if (result.isSuccess)
+              sys.exit(0)
+            else {
+              exitSys(log, "Error while processing institutions", 4)
+            }
+          })
+        } else {
+          exitSys(log, "Error creating institutions schema", 3)
+        })
+    })
+  }
+
+}
+