Using akka extensions mechanism

Author: t3hnar

src/main/resources/reference.conf

@@ -5,6 +5,7 @@
 # This file has all the default settings, so all these could be removed with no visible effect.
 # Modify as needed.
 akka {
+  extensions = ["com.thenewmotion.akka.http.HttpExtension$"]
   http {
     system-name = http                      # name of the system used in servlet
     endpoints-path = endpoints              # path of the actor to use for receiving and adding endpoints, call 'actorFor(/user/endpoints)'

src/main/scala/com/thenewmotion/akka/http/ActorHttpSystem.scala

@@ -2,20 +2,16 @@ package com.thenewmotion.akka.http
 
 import javax.servlet.http.{HttpServletResponse, HttpServletRequest, HttpServlet}
 import akka.actor.{ActorSystem, Props}
-import Http._
 
 
 class AkkaHttpServlet extends HttpServlet {
 
-  private[http] var _actorSystem: Option[ActorHttpSystem] = None
+  private[http] var _actorSystem: Option[ActorSystem] = None
 
   override def init() {
     super.init()
-
-    val system = ActorHttpSystem()
-    system.actorOf(Props[EndpointsActor], system.endpointsPath)
-    _actorSystem = Some(system)
-    onSystemInit(system)
+    _actorSystem = Some(HttpSystem())
+    onSystemInit(_actorSystem.get)
   }
 
   override def destroy() {
@@ -29,8 +25,8 @@ class AkkaHttpServlet extends HttpServlet {
     _actorSystem = None
   }
 
-  def onSystemInit(system: ActorHttpSystem) {}
-  def onSystemDestroy(system: ActorHttpSystem) {}
+  def onSystemInit(system: ActorSystem) {}
+  def onSystemDestroy(system: ActorSystem) {}
 
   override def doPost(req: HttpServletRequest, res: HttpServletResponse) {doActor(req, res)}
   override def doPut(req: HttpServletRequest, res: HttpServletResponse) {doActor(req, res)}
@@ -45,7 +41,7 @@ class AkkaHttpServlet extends HttpServlet {
     val actor = system.actorOf(props)
 
     val asyncContext = req.startAsync()
-    asyncContext.setTimeout(system.asyncTimeout)
+    asyncContext.setTimeout(HttpExtension(system).asyncTimeout)
     asyncContext.addListener(new Listener(actor, system))
 
     actor ! asyncContext

src/main/scala/com/thenewmotion/akka/http/AkkaHttpServlet.scala

@@ -5,7 +5,6 @@ import akka.actor._
 import akka.util.duration._
 import javax.servlet.{ServletRequest, ServletResponse, AsyncContext}
 import Endpoints._
-import Http._
 import Async._
 
 /**
@@ -15,18 +14,18 @@ class AsyncActor extends Actor with LoggingFSM[State, Data] {
 
   implicit def res2HttpRes(res: ServletResponse) = res.asInstanceOf[HttpServletResponse]
   implicit def req2HttpReq(req: ServletRequest) = req.asInstanceOf[HttpServletRequest]
-  val endpointTimeout = context.system.http.endpointRetrievalTimeout
+  val endpointTimeout = HttpExtension(context.system).endpointRetrievalTimeout
 
   startWith(Idle, Empty)
 
   when(Idle) {
     case Event(async: AsyncContext, Empty) =>
       val url = async.getRequest.getPathInfo
-      log.debug("Started async for '{}'", url)
-      context.system.http.endpoints ! Find(url)
-      goto(Started) using Context(async, url)
+      log.debug("AboutToProcess$ async for '{}'", url)
+      HttpExtension(context.system).endpoints ! Find(url)
+      goto(AboutToProcess) using Context(async, url)
   }
-  when(Started, endpointTimeout millis) {
+  when(AboutToProcess, endpointTimeout millis) {
     case Event(Found(EndpointFunc(func)), ctx@Context(_, url)) =>
       log.debug("Processing async for '{}'", url)
       safeProcess(func, ctx)
@@ -37,9 +36,9 @@ class AsyncActor extends Actor with LoggingFSM[State, Data] {
       log.debug("No endpoint received within {} millis for '{}'", endpointTimeout, url)
       safeProcess(NotFound, ctx)
   }
-  when(Completing) {
+  when(AboutToComplete) {
     case Event(Complete(completing), ctx@Context(async, url)) =>
-      log.debug("Completing async for '{}'", url)
+      log.debug("AboutToComplete$ async for '{}'", url)
 
       def doComplete(callback: Callback) {
         val success = try {
@@ -93,25 +92,24 @@ class AsyncActor extends Actor with LoggingFSM[State, Data] {
     catch InternalErrorOnException(async.url)
 
     //we want receive different messages before responding, for example 'Timeout'
-    goto(Completing) using async
+    goto(AboutToComplete) using async
   }
 
   def safeProcess(actor: ActorRef, async: Context): State = {
     try actor ! async.context.getRequest
     catch InternalErrorOnException(async.url)
 
     //actor should respond with Complete(..) message
-    goto(Completing) using async
+    goto(AboutToComplete) using async
   }
 }
 
 
 object Async {
   sealed trait State
   case object Idle extends State
-  case object Started extends State
-  case object ProcessingRequest extends State
-  case object Completing extends State
+  case object AboutToProcess extends State
+  case object AboutToComplete extends State
 
   sealed trait Data
   case class Context(context: AsyncContext, url: String) extends Data

src/main/scala/com/thenewmotion/akka/http/AsyncActor.scala

@@ -0,0 +1,35 @@
+package com.thenewmotion.akka.http
+
+import com.typesafe.config.ConfigFactory
+import akka.actor._
+
+
+/**
+ * @author Yaroslav Klymko
+ */
+
+object HttpSystem {
+  def apply(): ActorSystem = {
+    val name = ConfigFactory.load().getString("akka.http.system-name")
+    val system = ActorSystem(name)
+    system.actorOf(Props[EndpointsActor], HttpExtension(system).endpointsName)
+    system
+  }
+}
+
+class HttpExtension(val system: ActorSystem) extends Extension {
+  private val config = system.settings.config
+  import config._
+
+  val asyncTimeout: Long = system.settings.config.getLong("akka.http.timeout")
+  val endpointsName: String = getString("akka.http.endpoints-path")
+  def endpoints: ActorRef = system.actorFor("/user/" + endpointsName)
+  val endpointRetrievalTimeout: Long = getLong("akka.http.endpoint-retrieval-timeout")
+  val expiredHeader: (String, String) =
+    getString("akka.http.expired-header-name") -> getString("akka.http.expired-header-value")
+}
+
+object HttpExtension extends ExtensionId[HttpExtension] with ExtensionIdProvider {
+  def lookup() = HttpExtension
+  def createExtension(system: ExtendedActorSystem) = new HttpExtension(system)
+}

src/main/scala/com/thenewmotion/akka/http/Http.scala

@@ -1,13 +1,13 @@
 package com.thenewmotion.akka.http
 
-import akka.actor.ActorRef
 import javax.servlet.{AsyncEvent, AsyncListener}
 import javax.servlet.http.HttpServletResponse
+import akka.actor.{ActorSystem, ActorRef}
 
 /**
  * @author Yaroslav Klymko
  */
-class Listener(actor: ActorRef, system: ActorHttpSystem) extends AsyncListener {
+class Listener(actor: ActorRef, system: ActorSystem) extends AsyncListener {
 
   import Listener._
 
@@ -28,7 +28,7 @@ class Listener(actor: ActorRef, system: ActorHttpSystem) extends AsyncListener {
     val asyncContext = event.getAsyncContext
 
     val res = asyncContext.getResponse.asInstanceOf[HttpServletResponse]
-    val (name, value) = system.expiredHeader()
+    val (name, value) = HttpExtension(system).expiredHeader
     res.addHeader(name, value)
     asyncContext.complete()
   }

src/main/scala/com/thenewmotion/akka/http/HttpSystem.scala

@@ -1,14 +1,15 @@
 package com.thenewmotion.akka.http
 
 import Endpoints._
+import akka.actor.ActorSystem
 
 /**
  * @author Yaroslav Klymko
  */
 abstract class StaticAkkaHttpServlet extends AkkaHttpServlet {
 
-  override def onSystemInit(system: ActorHttpSystem) {
-    system.endpoints ! Attach("static", providers)
+  override def onSystemInit(system: ActorSystem) {
+    HttpExtension(system).endpoints ! Attach("static", providers)
   }
 
   def providers: Provider

src/main/scala/com/thenewmotion/akka/http/Listener.scala

@@ -32,8 +32,7 @@ class AkkaHttpServletSpec extends SpecificationWithJUnit {
 
     "shutdown ActorSystem on destroy" >> {
       servlet._actorSystem must beSome
-      val httpSystem = servlet._actorSystem.get
-      val system = httpSystem.system
+      val system = servlet._actorSystem.get
       servlet.destroy()
       system.awaitTermination()
       system.isTerminated must beTrue

src/main/scala/com/thenewmotion/akka/http/StaticAkkaHttpServlet.scala

@@ -39,7 +39,7 @@ class AsyncActorSpec extends SpecificationWithJUnit with Mockito {
       actorRef.stateName mustEqual Idle
       actorRef.stateData mustEqual Empty
       actorRef ! asyncContext
-      actorRef.stateName mustEqual Started
+      actorRef.stateName mustEqual AboutToProcess
       actorRef.stateData mustEqual Context(asyncContext, "/test")
     }
 

src/test/scala/com/thenewmotion/akka/http/AkkaHttpServletSpec.scala

@@ -0,0 +1,23 @@
+package com.thenewmotion.akka.http
+
+import org.specs2.mutable.SpecificationWithJUnit
+import akka.actor.ActorSystem
+
+/**
+ * @author Yaroslav Klymko
+ */
+class HttpExtensionSpec extends SpecificationWithJUnit {
+
+  "HttpExtension" should {
+    "load on startup" >> {
+      println(">>")
+      println("system >> ")
+      val system = ActorSystem()
+      println("system << ")
+      println("Extension >> ")
+      val extension = HttpExtension(system)
+      println("<<")
+      todo
+    }
+  }
+}