Refactored Retrieve to use Actors

Author: almacken

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

@@ -0,0 +1,39 @@
+package de.upb.cs.swt.delphi.webapi
+
+import akka.actor.{Actor, ActorLogging, PoisonPill, Props, ReceiveTimeout}
+import com.sksamuel.elastic4s.IndexAndType
+import com.sksamuel.elastic4s.http.ElasticDsl._
+import com.sksamuel.elastic4s.http.HttpClient
+import de.upb.cs.swt.delphi.webapi.ElasticActor.GetSource
+
+import scala.concurrent.duration._
+
+class ElasticActor(configuration: Configuration) extends Actor with ActorLogging{
+
+  val client = HttpClient(configuration.elasticsearchClientUri)
+
+  override def preStart(): Unit = log.info("Search actor started")
+  override def postStop(): Unit = log.info("Search actor shut down")
+  context.setReceiveTimeout(2 seconds)
+
+  override def receive = {
+    case GetSource(id, index) => {
+      log.info("Executing get on entry {}", id)
+      def source = client.execute{
+          get(id).from(index)
+        }.await match {
+          case Right(res) => res.body.get
+          case Left(_) => Option.empty
+        }
+      sender().tell(source, context.self)
+      context.stop(self)
+    }
+    case ReceiveTimeout => context.stop(self)
+  }
+}
+
+object ElasticActor{
+  def props(configuration: Configuration) : Props = Props(new ElasticActor(configuration))
+
+  final case class GetSource(id: String, index: IndexAndType)
+}

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

@@ -0,0 +1,28 @@
+package de.upb.cs.swt.delphi.webapi
+
+import akka.actor.{Actor, ActorLogging, Props}
+import com.sksamuel.elastic4s.http.ElasticDsl._
+import de.upb.cs.swt.delphi.webapi.ElasticActorManager.Retrieve
+import de.upb.cs.swt.delphi.webapi.ElasticActor.GetSource
+
+class ElasticActorManager(configuration: Configuration) extends Actor with ActorLogging{
+
+  private val index = "delphi" / "project"
+
+  override def preStart(): Unit = log.info("Actor manager started")
+  override def postStop(): Unit = log.info("Actor manager shut down")
+
+  override def receive = {
+    case Retrieve(id) => {
+      log.info("Creating actor to search for entry {}", id)
+      val retrieveActor = context.actorOf(ElasticActor.props(configuration))
+      retrieveActor forward GetSource(id, index)
+    }
+  }
+}
+
+object ElasticActorManager{
+  def props(configuration: Configuration) : Props = Props(new ElasticActorManager(configuration))
+
+  final case class Retrieve(id: String)
+}

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

@@ -1,14 +1,25 @@
 package de.upb.cs.swt.delphi.webapi
 
+import java.util.concurrent.TimeUnit
+
+import akka.actor.ActorSystem
 import akka.http.scaladsl.server.HttpApp
+import akka.pattern.ask
+import akka.util.Timeout
 import de.upb.cs.swt.delphi.featuredefinitions.FeatureListMapping
+import de.upb.cs.swt.delphi.webapi.ElasticActorManager.Retrieve
 import spray.json._
 
 /**
   * Web server configuration for Delphi web API.
   */
 object Server extends HttpApp with JsonSupport {
 
+  private val configuration = new Configuration()
+  private val system = ActorSystem("delphi-webapi")
+  private val actorManager = system.actorOf(ElasticActorManager.props(configuration))
+  implicit val timeout = Timeout(5, TimeUnit.SECONDS)
+
    override def routes =
       path("version") { version } ~
         path("features") { features } ~
@@ -35,7 +46,7 @@ object Server extends HttpApp with JsonSupport {
   def retrieve(identifier: String) = {
     get {
       complete(
-        ElasticClient.getSource(identifier)
+        (actorManager ? Retrieve(identifier)).mapTo[String]
       )
     }
   }
@@ -51,6 +62,7 @@ object Server extends HttpApp with JsonSupport {
   def main(args: Array[String]): Unit = {
     val configuration = new Configuration()
     Server.startServer(configuration.bindHost, configuration.bindPort)
+    system.terminate()
   }