Added Enqueue route, using a priority mailbox

Author: almacken

src/main/resources/application.conf

@@ -0,0 +1,9 @@
+es-priority-mailbox {
+  mailbox-type = "de.upb.cs.swt.delphi.webapi.ElasticPriorityMailbox"
+}
+
+akka.actor.deployment {
+  /espriomailboxactor {
+    mailbox = es-priority-mailbox
+  }
+}

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

@@ -0,0 +1,8 @@
+package de.upb.cs.swt.delphi.webapi
+
+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)
+}

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

@@ -1,6 +1,6 @@
 package de.upb.cs.swt.delphi.webapi
 
-import akka.actor.{Actor, ActorLogging, PoisonPill, Props, ReceiveTimeout}
+import akka.actor.{Actor, ActorLogging, Props, ReceiveTimeout}
 import com.sksamuel.elastic4s.IndexAndType
 import com.sksamuel.elastic4s.http.ElasticDsl._
 import com.sksamuel.elastic4s.http.HttpClient

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

@@ -2,7 +2,7 @@ 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.ElasticActorManager.{Enqueue, Retrieve}
 import de.upb.cs.swt.delphi.webapi.ElasticActor.GetSource
 
 class ElasticActorManager(configuration: Configuration) extends Actor with ActorLogging{
@@ -13,16 +13,21 @@ class ElasticActorManager(configuration: Configuration) extends Actor with Actor
   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)
-    }
+    case Retrieve(id) => getSource(id)
+    case Enqueue(id) =>  getSource(id)
+  }
+
+  private def getSource(id: String) = {
+    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))
+    .withMailbox("es-priority-mailbox")
 
   final case class Retrieve(id: String)
+  final case class Enqueue(id: String)
 }

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

@@ -0,0 +1,14 @@
+package de.upb.cs.swt.delphi.webapi
+
+import akka.actor.ActorSystem
+import akka.dispatch.{PriorityGenerator, UnboundedStablePriorityMailbox}
+import de.upb.cs.swt.delphi.webapi.ElasticActorManager.{Enqueue, Retrieve}
+import com.typesafe.config.Config
+
+class ElasticPriorityMailbox (settings: ActorSystem.Settings, config: Config)
+  extends UnboundedStablePriorityMailbox(
+    PriorityGenerator{
+      case Retrieve(_) => 5
+      case Enqueue(_) => 1
+      case _ => 2
+    })

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

@@ -7,13 +7,13 @@ 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 de.upb.cs.swt.delphi.webapi.ElasticActorManager.{Enqueue, Retrieve}
 import spray.json._
 
 /**
   * Web server configuration for Delphi web API.
   */
-object Server extends HttpApp with JsonSupport {
+object Server extends HttpApp with JsonSupport with AppLogging {
 
   private val configuration = new Configuration()
   private val system = ActorSystem("delphi-webapi")
@@ -24,7 +24,8 @@ object Server extends HttpApp with JsonSupport {
       path("version") { version } ~
         path("features") { features } ~
         pathPrefix("search" / Remaining) { query => search(query) } ~
-        pathPrefix("retrieve" / Remaining) { identifier => retrieve(identifier) }
+        pathPrefix("retrieve" / Remaining) { identifier => retrieve(identifier) } ~
+        pathPrefix("enqueue" / Remaining) { identifier => enqueue(identifier) }
 
 
   private def version = {
@@ -51,6 +52,16 @@ object Server extends HttpApp with JsonSupport {
     }
   }
 
+  def enqueue(identifier: String) = {
+    get {
+      pass {    //TODO: Require authorization here
+        complete(
+          (actorManager ? Enqueue(identifier)).mapTo[String]
+        )
+      }
+    }
+  }
+
   def search(query: String) = {
     get {
       complete {