Merge branch 'feature/adapt-swagger-model' into feature/instanceregistry

Author: Johannes Duesing

src/main/scala/de/upb/cs/swt/delphi/instancemanagement/Instance.scala

@@ -5,7 +5,7 @@ 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 write(compType : InstanceEnums.ComponentType) = JsString(compType.toString)
 
     def read(value: JsValue) = value match {
       case JsString(s) => s match {
@@ -24,11 +24,11 @@ trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
 
 final case class Instance (
                             iD: Option[Long],
-                            iP: Option[String],
-                            portnumber: Option[Long],
-                            name: Option[String],
+                            host: String,
+                            portnumber: Int,
+                            name: String,
                             /* Component Type */
-                            componentType: Option[InstanceEnums.ComponentType]
+                            componentType: InstanceEnums.ComponentType
                           )
 
 object InstanceEnums {
@@ -42,4 +42,4 @@ object InstanceEnums {
     val ElasticSearch = Value("ElasticSearch")
   }
 
-}
+}

src/main/scala/de/upb/cs/swt/delphi/instancemanagement/InstanceRegistry.scala

@@ -2,24 +2,25 @@ package de.upb.cs.swt.delphi.instancemanagement
 
 import java.net.InetAddress
 
+import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.marshalling.Marshal
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.unmarshalling.Unmarshal
+import akka.stream.ActorMaterializer
 import de.upb.cs.swt.delphi.instancemanagement.InstanceEnums.ComponentType
 import de.upb.cs.swt.delphi.webapi.{AppLogging, Configuration, Server}
 
-
-import scala.concurrent.{Await, Future}
+import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.concurrent.duration.Duration
 import scala.util.{Failure, Success, Try}
 
 object InstanceRegistry extends JsonSupport with AppLogging
 {
 
- implicit val system = Server.system
-  implicit val ec = system.dispatcher
-  implicit val materializer = Server.materializer
+  implicit val system : ActorSystem = Server.system
+  implicit val ec  : ExecutionContext = system.dispatcher
+  implicit val materializer : ActorMaterializer = Server.materializer
 
 
   def register(configuration: Configuration) : Try[Long] = {
@@ -59,8 +60,8 @@ object InstanceRegistry extends JsonSupport with AppLogging
         if(status == StatusCodes.OK) {
 
           Await.result(Unmarshal(response.entity).to[Instance] map {instance =>
-            val elasticIP = instance.iP
-            log.info(s"Instance Registry assigned ElasticSearch instance at ${elasticIP.getOrElse("None")}")
+            val elasticIP = instance.host
+            log.info(s"Instance Registry assigned ElasticSearch instance at $elasticIP")
             Success(instance)
           } recover {case ex =>
             log.warning(s"Failed to read response from Instance Registry, exception: $ex")
@@ -85,10 +86,10 @@ object InstanceRegistry extends JsonSupport with AppLogging
       if(configuration.elasticsearchInstance.iD.isEmpty) {
         Failure(new RuntimeException("Cannot post matching result to Instance Registry, assigned ElasticSearch instance has no ID."))
       } else {
-        val IdToPost = configuration.elasticsearchInstance.iD.get
+        val idToPost = configuration.elasticsearchInstance.iD.getOrElse(-1L)
         val request = HttpRequest(
           method = HttpMethods.POST,
-          configuration.instanceRegistryUri + s"/matchingResult?Id=$IdToPost&MatchingSuccessful=$isElasticSearchReachable")
+          configuration.instanceRegistryUri + s"/matchingResult?Id=$idToPost&MatchingSuccessful=$isElasticSearchReachable")
 
         Await.result(Http(system).singleRequest(request) map {response =>
           if(response.status == StatusCodes.OK){
@@ -114,7 +115,7 @@ object InstanceRegistry extends JsonSupport with AppLogging
     if(!configuration.usingInstanceRegistry){
       Failure(new RuntimeException("Cannot deregister from Instance Registry, no Instance Registry available."))
     } else {
-      val id : Long = configuration.assignedID.get
+      val id : Long = configuration.assignedID.getOrElse(-1L)
 
       val request = HttpRequest(method = HttpMethods.POST, configuration.instanceRegistryUri + s"/deregister?Id=$id")
 
@@ -144,5 +145,5 @@ object InstanceRegistry extends JsonSupport with AppLogging
 
 
   private def createInstance(id: Option[Long], controlPort : Int, name : String) : Instance =
-    Instance(id, Option(InetAddress.getLocalHost.getHostAddress), Option(controlPort), Option(name), Option(ComponentType.Crawler))
+    Instance(id, InetAddress.getLocalHost.getHostAddress, controlPort, name, ComponentType.WebApi)
 }

src/main/scala/de/upb/cs/swt/delphi/webapi/Configuration.scala

@@ -5,7 +5,7 @@ import com.sksamuel.elastic4s.http.ElasticDsl._
 import de.upb.cs.swt.delphi.instancemanagement.InstanceEnums.ComponentType
 import de.upb.cs.swt.delphi.instancemanagement.{Instance, InstanceRegistry}
 
-import scala.util.{Failure, Success}
+import scala.util.{Failure, Success, Try}
 
 /**
   * @author Ben Hermann
@@ -20,21 +20,22 @@ class Configuration(  //Server and Elasticsearch configuration
                    ) {
 
 
-  lazy val elasticsearchClientUri: ElasticsearchClientUri = ElasticsearchClientUri({
-    if(elasticsearchInstance.portnumber.isEmpty){
-      elasticsearchInstance.iP.getOrElse("elasticsearch://localhost:9200")
-    }else{
-      elasticsearchInstance.iP.getOrElse("elasticsearch://localhost") + ":" + elasticsearchInstance.portnumber.getOrElse("9200")
-    }
-  })
+  lazy val elasticsearchClientUri: ElasticsearchClientUri = ElasticsearchClientUri(
+    elasticsearchInstance.host + ":" + elasticsearchInstance.portnumber)
 
   lazy val elasticsearchInstance : Instance = InstanceRegistry.retrieveElasticSearchInstance(this) match {
     case Success(instance) => instance
-    case Failure(_) => Instance(None, Some(sys.env.getOrElse("DELPHI_ELASTIC_URI","elasticsearch://localhost:9200")), None, Some("Default ElasticSearch instance"), Some(ComponentType.ElasticSearch) )
+    case Failure(_) => Instance(
+      None,
+      fallbackElasticSearchHost,
+      fallbackElasticSearchPort,
+      "Default ElasticSearch instance",
+      ComponentType.ElasticSearch)
   }
-
+  val defaultElasticSearchPort : Int = 9200
+  val defaultElasticSearchHost : String = "elasticsearch://localhost"
   val instanceName = "MyWebApiInstance"
-  val instanceRegistryUri : String = sys.env.getOrElse("DELPHI_IR_URI", "http://localhost:8085")
+  val instanceRegistryUri : String = sys.env.getOrElse("DELPHI_IR_URI", "http://localhost:8087")
   lazy val usingInstanceRegistry : Boolean = assignedID match {
     case Some(_) => true
     case None => false
@@ -43,5 +44,29 @@ class Configuration(  //Server and Elasticsearch configuration
     case Success(id) => Some(id)
     case Failure(_) => None
   }
+  lazy val fallbackElasticSearchPort : Int = sys.env.get("DELPHI_ELASTIC_URI") match {
+    case Some(hostString) => if(hostString.count(c => c == ':') == 3){
+      Try(hostString.split(":")(2).toInt) match {
+        case Success(port) => port
+        case Failure(_) => defaultElasticSearchPort
+      }
+    } else {
+      defaultElasticSearchPort
+    }
+    case None => defaultElasticSearchPort
+  }
+
+  lazy val fallbackElasticSearchHost : String = sys.env.get("DELPHI_ELASTIC_URI") match {
+    case Some(hostString) =>
+      if(hostString.count(c => c == ':') == 2){
+        hostString.substring(0,hostString.lastIndexOf(":"))
+      } else {
+        defaultElasticSearchHost
+      }
+    case None => defaultElasticSearchHost
+
+  }
 
 }
+
+