Configure cluster receptionist

Author: jmarin

cluster/src/main/resources/application.conf

@@ -39,7 +39,7 @@ akka {
 
   }
 
-  extensions = ["de.heikoseeberger.constructr.ConstructrExtension"]
+  extensions = ["de.heikoseeberger.constructr.ConstructrExtension", "akka.cluster.client.ClusterClientReceptionist"]
 
 }
 
@@ -64,6 +64,7 @@ clustering {
   name = "hmda"
   name = ${?HMDA_CLUSTER_NAME}
   ip = "127.0.0.1"
+  ip = ${?HMDA_CLUSTER_IP}
   port = 0
   port = ${?APP_PORT}
 }

loader/src/main/resources/application.conf

@@ -1,12 +1,17 @@
 akka {
-  loglevel = "INFO"
+  loglevel = "DEBUG"
   loglevel = ${?HMDA_LOGLEVEL}
   actor.warn-about-java-serializer-usage = off
+  actor.provider = cluster
+
+  extensions = ["akka.cluster.client.ClusterClientReceptionist"]
 }
 
 hmda {
+  clusterName = "hmda"
   adminUrl = "http://0.0.0.0:8081"
   adminUrl = ${?HMDA_HTTP_ADMIN_URL}
+  actorTimeout = 5
   httpTimeout = 5
   panel {
     tcp {
@@ -16,5 +21,11 @@ hmda {
       port = ${?HMDA_PANEL_LOADER_PORT}
     }
   }
+  lar {
+    host = "127.0.0.1"
+    host = ${?HMDA_CLUSTER_HOST}
+    port = 0
+    port = ${?HMDA_CLUSTER_PORT}
+  }
 }
 

loader/src/main/scala/hmda/loader/lar/HmdaLarLoader.scala

@@ -0,0 +1,60 @@
+package hmda.loader.lar
+
+import akka.actor.{ActorPath, ActorRef, ActorSystem}
+import akka.pattern.ask
+import akka.cluster.client.{ClusterClient, ClusterClientSettings}
+import akka.util.Timeout
+import com.typesafe.config.ConfigFactory
+import hmda.model.fi.SubmissionId
+import hmda.persistence.HmdaSupervisor.{FindHmdaFiling, FindProcessingActor, FindSubmissions}
+import hmda.persistence.institutions.SubmissionPersistence
+import hmda.persistence.processing.SubmissionManager
+import scala.concurrent.duration._
+
+object HmdaLarLoader extends App {
+
+  val config = ConfigFactory.load()
+
+  implicit val system = ActorSystem("hmda-cluster-client")
+  implicit val ec = system.dispatcher
+
+  val hmdaClusterName = config.getString("hmda.clusterName")
+  val hmdaClusterIP = config.getString("hmda.lar.host")
+  val hmdaClusterPort = config.getInt("hmda.lar.port")
+  val actorTimeout = config.getInt("hmda.actorTimeout")
+
+  implicit val timeout = Timeout(actorTimeout.seconds)
+
+  val initialContacts = Set(
+    ActorPath.fromString(s"akka.tcp://$hmdaClusterName@$hmdaClusterIP:$hmdaClusterPort/system/receptionist")
+  )
+
+  println(initialContacts)
+
+  val settings = ClusterClientSettings(system)
+    .withInitialContacts(initialContacts)
+
+  val clusterClient = system.actorOf(ClusterClient.props(settings), "hmda-lar-loader")
+
+  val institutionId = "institutionID"
+  val period = "2017"
+
+  val submissionId = SubmissionId(institutionId, period)
+
+  val message = FindProcessingActor(SubmissionManager.name, submissionId)
+
+  val fProcessingActor = (clusterClient ? ClusterClient.Send("/user/supervisor/singleton", message, localAffinity = true)).mapTo[ActorRef]
+  val fSubmissionsActor = (clusterClient ? ClusterClient
+    .Send("/user/supervisor/singleton",
+      FindSubmissions(SubmissionPersistence.name, institutionId, period),
+      localAffinity = true)).mapTo[ActorRef]
+
+  (clusterClient ? ClusterClient.Send("/user/supervisor/singleton", FindHmdaFiling(period), localAffinity = true)).mapTo[ActorRef]
+  //(clusterClient ? ClusterClient.Send("/user/query-supervisor", FindHmdaFilingView(period), localAffinity = true)).mapTo[ActorRef]
+
+
+
+
+  private def uploadFile(): Unit = ???
+
+}

persistence/src/main/scala/hmda/persistence/HmdaSupervisor.scala

@@ -1,6 +1,7 @@
 package hmda.persistence
 
 import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.cluster.client.ClusterClientReceptionist
 import hmda.model.fi.SubmissionId
 import hmda.persistence.institutions.{ FilingPersistence, InstitutionPersistence, SubmissionPersistence }
 import hmda.persistence.model.HmdaSupervisorActor
@@ -25,6 +26,8 @@ class HmdaSupervisor(validationStats: ActorRef) extends HmdaSupervisorActor {
 
   import HmdaSupervisor._
 
+  ClusterClientReceptionist(context.system).registerService(self)
+
   override def receive: Receive = super.receive orElse {
 
     case FindHmdaFiling(filingPeriod) =>

query/src/main/scala/hmda/query/HmdaQuerySupervisor.scala

@@ -1,12 +1,14 @@
 package hmda.query
 
 import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.cluster.client.ClusterClientReceptionist
 import akka.util.Timeout
 import hmda.persistence.model.HmdaSupervisorActor
 import hmda.query.view.filing.HmdaFilingView
 import hmda.query.view.institutions.InstitutionView
 import hmda.persistence.PersistenceConfig._
 import hmda.persistence.messages.CommonMessages._
+
 import scala.concurrent.duration._
 
 object HmdaQuerySupervisor {
@@ -28,6 +30,8 @@ class HmdaQuerySupervisor extends HmdaSupervisorActor {
   implicit val timeout = Timeout(duration.seconds)
   implicit val ec = context.dispatcher
 
+  ClusterClientReceptionist(context.system).registerService(self)
+
   override def receive: Receive = super.receive orElse {
     case FindHmdaFilingView(period) =>
       sender() ! findHmdaFilingView(period)