Add TCP panel loader

Author: jmarin

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

@@ -28,6 +28,15 @@ hmda {
     timeout = ${?HMDA_HTTP_TIMEOUT}
   }
   isDemo = true
+  panel {
+    tcp {
+      host = "127.0.0.1"
+      host = ${?HMDA_PANEL_LOADER_HOST}
+      port = "8888"
+      port = ${?HMDA_PANEL_LOADER_PORT}
+      timeout = 5
+    }
+  }
   actor-flow-parallelism = 4
 }
 

api/src/main/resources/application.conf

@@ -35,6 +35,8 @@ hmda {
       host = ${?HMDA_PANEL_LOADER_HOST}
       port = "8888"
       port = ${?HMDA_PANEL_LOADER_PORT}
+      timeout = 5
+      parallelism = 5
     }
   }
 }

api/src/main/scala/hmda/api/tcp/TcpApi.scala

@@ -2,13 +2,13 @@ package hmda.api.tcp
 
 import java.net.InetSocketAddress
 
-import akka.actor.{Actor, ActorSystem, Status}
+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}
+import scala.concurrent.{ ExecutionContext, Future }
 
 abstract class TcpApi extends HmdaActor {
 

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

@@ -1,15 +1,26 @@
 package hmda.api.tcp.admin
 
 import akka.NotUsed
-import akka.pattern.pipe
-import akka.actor.{ActorRef, ActorSystem}
+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
+import akka.stream.scaladsl.{ Flow, Tcp }
+import akka.util.{ ByteString, Timeout }
 import hmda.api.tcp.TcpApi
 import hmda.api.util.FlowUtils
+import hmda.model.institution.Institution
+import hmda.parser.fi.InstitutionParser
+import hmda.persistence.institutions.InstitutionPersistence
+import hmda.persistence.institutions.InstitutionPersistence.CreateInstitution
+import hmda.persistence.model.HmdaSupervisorActor.FindActorByName
+import scala.concurrent.{ ExecutionContext, Future }
+import scala.concurrent.duration._
 
-import scala.concurrent.{ExecutionContext, Future}
+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"
@@ -20,12 +31,20 @@ class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils
 
   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 fInstitutionsActor = (supervisor ? FindActorByName(InstitutionPersistence.name)).mapTo[ActorRef]
 
   val tcpHandler: Flow[ByteString, ByteString, NotUsed] =
     Flow[ByteString]
-    .map{ e => println(e); e}
-
+      .via(framing)
+      .drop(1)
+      .via(byte2StringFlow)
+      .map(x => InstitutionParser(x))
+      .mapAsync(parallelism = buffer)(i => createInstitution(fInstitutionsActor, i))
+      .map(e => ByteString(e.toString))
 
   override val tcp: Future[Tcp.ServerBinding] = Tcp().bindAndHandle(
     tcpHandler,
@@ -34,4 +53,12 @@ class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils
   )
 
   tcp pipeTo self
+
+  private def createInstitution(fActor: Future[ActorRef], i: Institution): Future[Institution] = {
+    for {
+      actor <- fActor
+      i <- (actor ? CreateInstitution(i)).mapTo[Option[Institution]].map(i => i.getOrElse(Institution.empty))
+    } yield i
+  }
+
 }

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
@@ -85,6 +86,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

loader/src/main/resources/application.conf

@@ -8,5 +8,13 @@ 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}
+    }
+  }
 }
 

loader/src/main/scala/hmda/loader/http/InstitutionFlowUtils.scala

@@ -1,18 +1,16 @@
-package hmda.panel
+package hmda.loader.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 akka.stream.scaladsl.{ FileIO, Sink, Tcp }
 import org.slf4j.LoggerFactory
 import hmda.api.util.FlowUtils
+
 import scala.concurrent.duration._
 
-object PanelCsvLoader extends FlowUtils with InstitutionFlowUtils {
+object PanelCsvLoader extends FlowUtils {
   val httpTimeout = config.getInt("hmda.httpTimeout")
   val adminUrl = config.getString("hmda.adminUrl")
   val url = s"$adminUrl/institutions"
@@ -22,6 +20,9 @@ object PanelCsvLoader extends FlowUtils with InstitutionFlowUtils {
   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) {
@@ -33,34 +34,15 @@ object PanelCsvLoader extends FlowUtils with InstitutionFlowUtils {
       exitSys(log, "File does not exist", 2)
     }
 
-    sendGetRequest("delete", url).map(_ => {
-      val response = sendGetRequest("create", url)
+    val connectionFlow = Tcp().outgoingConnection(host, port)
 
-      val source = FileIO.fromPath(file.toPath)
+    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)))
+    source
+      .via(connectionFlow)
+      .runWith(Sink.last)
+      .onComplete(_ => system.terminate())
 
-          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)
-        })
-    })
   }
-
 }