Improved priority handling

Author: almacken

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

@@ -4,12 +4,12 @@ import akka.actor.{Actor, ActorLogging, Props}
 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 de.upb.cs.swt.delphi.webapi.ElasticActorManager.{Enqueue, Retrieve}
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
-class ElasticActor(configuration: Configuration) extends Actor with ActorLogging{
+class ElasticActor(configuration: Configuration, index: IndexAndType) extends Actor with ActorLogging{
 
   implicit val executionContext: ExecutionContext = context.system.dispatchers.lookup("elasticsearch-handling-dispatcher")
   val client = HttpClient(configuration.elasticsearchClientUri)
@@ -19,21 +19,23 @@ class ElasticActor(configuration: Configuration) extends Actor with ActorLogging
   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)
+    case Enqueue(id) => getSource(id)
+    case Retrieve(id) => getSource(id)
+  }
+
+  private def getSource(id: String) = {
+    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)
   }
 }
 
 object ElasticActor{
-  def props(configuration: Configuration) : Props = Props(new ElasticActor(configuration))
-
-  final case class GetSource(id: String, index: IndexAndType)
+  def props(configuration: Configuration, index: IndexAndType) : Props = Props(new ElasticActor(configuration, index))
+    .withMailbox("es-priority-mailbox")
 }

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

@@ -2,15 +2,14 @@ package de.upb.cs.swt.delphi.webapi
 
 import akka.actor.{Actor, ActorLogging, Props, Terminated}
 import akka.routing.{ActorRefRoutee, RoundRobinRoutingLogic, Router}
-import de.upb.cs.swt.delphi.webapi.ElasticActorManager.{Enqueue, Retrieve}
-import de.upb.cs.swt.delphi.webapi.ElasticActor.GetSource
+import de.upb.cs.swt.delphi.webapi.ElasticActorManager.ElasticMessage
 
 class ElasticActorManager(configuration: Configuration) extends Actor with ActorLogging{
 
   private val index = configuration.esProjectIndex
   private var elasticRouter = {
     val routees = Vector.fill(configuration.elasticActorPoolSize) {
-      val r = context.actorOf(ElasticActor.props(configuration))
+      val r = context.actorOf(ElasticActor.props(configuration, index))
       context watch r
       ActorRefRoutee(r)
     }
@@ -21,20 +20,17 @@ class ElasticActorManager(configuration: Configuration) extends Actor with Actor
   override def postStop(): Unit = log.info("Actor manager shut down")
 
   override def receive = {
-    case Retrieve(id) => getSource(id)
-    case Enqueue(id) =>  getSource(id)
+    case em: ElasticMessage => {
+      log.info("Forwarding request {} to ElasticActor", em)
+      elasticRouter.route(em, sender())
+    }
     case Terminated(id) => {
       elasticRouter.removeRoutee(id)
-      val r = context.actorOf(ElasticActor.props(configuration))
+      val r = context.actorOf(ElasticActor.props(configuration, index))
       context watch r
       elasticRouter = elasticRouter.addRoutee(r)
     }
   }
-
-  private def getSource(id: String) = {
-    log.info("Forwarding search for entry {}", id)
-    elasticRouter.route(GetSource(id, index), sender())
-  }
 }
 
 object ElasticActorManager{