Merge pull request #19 from delphi-hub/feature/config_ip_for_IR

Webapp connection to instance registry

Author: bhermann

app/EagerLoaderModule.scala

@@ -0,0 +1,13 @@
+package app
+import com.google.inject.AbstractModule
+import services.StartUpService
+
+/**
+  * Run functions during request
+  */
+class EagerLoaderModule extends AbstractModule {
+  override def configure() = {
+    //startupservice will run during request
+    bind(classOf[StartUpService]).asEagerSingleton
+  }
+}

app/services/StartUpService.scala

@@ -0,0 +1,46 @@
+package services
+
+import java.util.concurrent.TimeUnit
+
+import javax.inject.{Singleton, _}
+import play.api.inject.ApplicationLifecycle
+import utils.Configuration
+import utils.instancemanagement.InstanceRegistry
+
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, Future}
+import scala.util.{Failure, Success}
+
+/**
+  * functions that will be run during request
+  */
+@Singleton
+class StartUpService @Inject()(appLifecycle: ApplicationLifecycle){
+
+  private val configuration = new Configuration()
+
+  /**
+    * Will register at the Instance Registry, get an matching WebApi instance and try to connect to it using the
+    * /version endpoint. If successful, it will post the matching result true to the IR, otherwise false.
+    */
+  def doStartUpChecks(): Unit = {
+    InstanceRegistry.getWebApiVersion(configuration) match {
+      case Success(_) => {
+        InstanceRegistry.sendWebApiMatchingResult(true, configuration)
+      }
+      case Failure(_) => {
+        InstanceRegistry.sendWebApiMatchingResult(false, configuration)
+        //Cannot connect to WebApi on startup, so stop execution
+        Await.ready(appLifecycle.stop(), Duration(5, TimeUnit.SECONDS))
+        System.exit(1)
+      }
+    }
+  }
+  
+  appLifecycle.addStopHook { () =>
+    InstanceRegistry.deregister(configuration)
+    Future.successful(())
+  }
+
+  doStartUpChecks()
+}

app/utils/AppLogging.scala

@@ -0,0 +1,8 @@
+package utils
+
+import akka.actor.{ActorSystem, ExtendedActorSystem}
+import akka.event.{BusLogging, LoggingAdapter}
+
+trait AppLogging {
+  def log(implicit system: ActorSystem): LoggingAdapter = new BusLogging(system.eventStream, this.getClass.getName, this.getClass, system.asInstanceOf[ExtendedActorSystem].logFilter)
+}

app/utils/BlockingHttpClient.scala

@@ -3,13 +3,15 @@ package utils
 
 import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
-import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse, Uri}
+import akka.http.scaladsl.model.{HttpEntity, HttpMethods, HttpRequest, HttpResponse, MediaTypes, Uri}
 import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
 import akka.util.ByteString
 
 import scala.concurrent.{Await, Future}
 import scala.concurrent.duration.Duration
 import scala.util.{Failure, Success, Try}
+import MediaTypes._
+
 
 /***
   * A blocking http client implemented using Akka HTTP
@@ -38,11 +40,44 @@ object BlockingHttpClient {
 
   }
 
+  // data parameter will be """{"name":"Hello"}"""
+  def doPost(uri: Uri, data: String) = {
+    implicit val system = ActorSystem()
+    implicit val executionContext = system.dispatcher
+    implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(system))
+    val bdata = ByteString(data)
+    try {
+      val req: Future[HttpResponse] = Http(system).singleRequest(HttpRequest(
+        method = HttpMethods.POST,
+        uri = uri,
+        entity = HttpEntity(`application/json`, bdata)
+      ))
+      Await.result(req, Duration.Inf)
+
+      val f = req.value.get.get.entity.dataBytes.runFold(ByteString(""))(_ ++ _)
+      Await.result(f, Duration.Inf)
+
+      Success(f.value.get.get.utf8String)
+    } catch  {
+      case e : Exception => Failure(e)
+    } finally {
+      system.terminate()
+      Await.result(system.whenTerminated, Duration.Inf)
+    }
+  }
+
   def executeGet(target: String, server: String) : Try[String] = {
 
     val uri = Uri(server)
     doGet(uri.withPath(uri.path + target))
 
   }
 
+  def executePost(target: String, server: String, data: String) : Try[String] = {
+
+    val uri = Uri(server)
+    doPost(uri.withPath(uri.path + target), data)
+
+  }
+
 }

app/utils/Configuration.scala

@@ -0,0 +1,60 @@
+package utils
+
+import com.typesafe.config.ConfigFactory
+import utils.instancemanagement.InstanceEnums.ComponentType
+import utils.instancemanagement.{Instance, InstanceRegistry}
+
+import scala.util.{Failure, Success, Try}
+
+class Configuration(val bindPort: Int = ConfigFactory.load().getInt("app.portWebapp")) {
+
+  val defaultWebApiPort : Int = ConfigFactory.load().getInt("webapi.port")
+  val defaultWebApiHost : String = ConfigFactory.load().getString("webapi.host")
+  val instanceName = "WebAppInstance"
+  val instanceRegistryUri : String = sys.env.getOrElse("DELPHI_WEBAPI_URI", ConfigFactory.load().getString("instance.registry.path"))
+
+  lazy val webApiUri:String =  webApiInstance.host + ":" + webApiInstance.portNumber
+
+  lazy val webApiInstance : Instance = InstanceRegistry.retrieveWebApiInstance(this) match {
+    case Success(instance) => instance
+    case Failure(_) => Instance(
+      None,
+      fallbackWebApiHost,
+      fallbackWebApiPort,
+      "Default WebApi instance",
+      ComponentType.WebApi)
+
+  }
+
+  lazy val usingInstanceRegistry : Boolean = assignedID match {
+    case Some(_) => true
+    case None => false
+  }
+  lazy val assignedID : Option[Long] = InstanceRegistry.register(this) match {
+    case Success(id) => Some(id)
+    case Failure(_) => None
+  }
+
+  lazy val fallbackWebApiPort : Int = sys.env.get("DELPHI_WEBAPI_URI") match {
+    case Some(hostString) => if(hostString.count(c => c == ':') == 2){
+      Try(hostString.split(":")(2).toInt) match {
+        case Success(port) => port
+        case Failure(_) => defaultWebApiPort
+      }
+    } else {
+      defaultWebApiPort
+    }
+    case None => defaultWebApiPort
+  }
+
+  lazy val fallbackWebApiHost : String = sys.env.get("DELPHI_WEBAPI_URI") match {
+    case Some(hostString) =>
+      if(hostString.count(c => c == ':') == 2){
+        hostString.substring(0,hostString.lastIndexOf(":"))
+      } else {
+        defaultWebApiHost
+      }
+    case None => defaultWebApiHost
+
+  }
+}

app/utils/JsonSupport.scala

@@ -0,0 +1,6 @@
+package utils
+
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
+import spray.json._
+
+trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol

app/utils/instancemanagement/Instance.scala

@@ -0,0 +1,46 @@
+package utils.instancemanagement
+
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
+import spray.json.{DefaultJsonProtocol, JsString, JsValue, JsonFormat}
+
+trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
+  implicit val componentTypeFormat = new JsonFormat[InstanceEnums.ComponentType] {
+    def write(compType : InstanceEnums.ComponentType) = JsString(compType.toString)
+
+    def read(value: JsValue) = value match {
+      case JsString(s) => s match {
+        case "Crawler" => InstanceEnums.ComponentType.Crawler
+        case "WebApi" => InstanceEnums.ComponentType.WebApi
+        case "WebApp" => InstanceEnums.ComponentType.WebApp
+        case "DelphiManagement" => InstanceEnums.ComponentType.DelphiManagement
+        case "ElasticSearch" => InstanceEnums.ComponentType.ElasticSearch
+        case x => throw new RuntimeException(s"Unexpected string value $x for component type.")
+      }
+      case y => throw new RuntimeException(s"Unexpected type $y while deserializing component type.")
+    }
+  }
+  implicit val instanceFormat = jsonFormat5(Instance)
+}
+
+final case class Instance (
+                            id: Option[Long],
+                            host: String,
+                            portNumber: Int,
+                            name: String,
+                            /* Component Type */
+                            componentType: InstanceEnums.ComponentType
+
+                          )
+
+object InstanceEnums {
+
+  type ComponentType = ComponentType.Value
+  object ComponentType extends Enumeration {
+    val Crawler = Value("Crawler")
+    val WebApi = Value("WebApi")
+    val WebApp = Value("WebApp")
+    val DelphiManagement = Value("DelphiManagement")
+    val ElasticSearch = Value("ElasticSearch")
+  }
+
+}