raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

make EventHandler non-global

Browse source 
- add Logging trait for nicer interface
- add EventHandlerLogging class for offering the nice interface from
  app.log
- add eventHandler instance to app and use that for all internal logging
  (this means that some places (dispatchers, remoting) were infiltrated
  by app just to do logging, but I think we'll need app in there soon
  enough for other reasons)
This commit is contained in:
Roland 2011-10-13 13:16:41 +02:00
parent e945dcfbe7
commit 32effb5083
3 changed files with 55 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

42
akka-http/src/main/scala/akka/http/JettyContinuation.scala
View file

@ -16,8 +16,8 @@ import akka.AkkaApplication
*/
trait JettyContinuation extends ContinuationListener {
import javax.servlet.http.HttpServletResponse
protected def application: AkkaApplication
def app: AkkaApplication
val builder: () ⇒ tAsyncRequestContext
val context: Option[tAsyncRequestContext] = Some(builder())
@ -35,7 +35,7 @@ trait JettyContinuation extends ContinuationListener {
// the fresh continuation (coming through getAsyncContinuation)
//
case (true, false, false) ⇒ {
continuation.setTimeout(application.MistSettings.DefaultTimeout)
continuation.setTimeout(app.MistSettings.DefaultTimeout)
continuation.addContinuationListener(this)
continuation.suspend
@ -47,7 +47,7 @@ trait JettyContinuation extends ContinuationListener {
//
case (true, true, false) ⇒ {
continuation.setTimeout(application.MistSettings.DefaultTimeout)
continuation.setTimeout(app.MistSettings.DefaultTimeout)
continuation.addContinuationListener(this)
Some(continuation)
@ -58,9 +58,9 @@ trait JettyContinuation extends ContinuationListener {
//
case (false, false, false) ⇒ {
continuation.setTimeout(continuation.getAttribute(application.MistSettings.TimeoutAttribute).asInstanceOf[Long])
continuation.setTimeout(continuation.getAttribute(app.MistSettings.TimeoutAttribute).asInstanceOf[Long])
continuation.suspend
continuation.removeAttribute(application.MistSettings.TimeoutAttribute)
continuation.removeAttribute(app.MistSettings.TimeoutAttribute)
None
}
@ -70,8 +70,8 @@ trait JettyContinuation extends ContinuationListener {
//
case (false, true, false) ⇒ {
continuation.setTimeout(continuation.getAttribute(application.MistSettings.TimeoutAttribute).asInstanceOf[Long])
continuation.removeAttribute(application.MistSettings.TimeoutAttribute)
continuation.setTimeout(continuation.getAttribute(app.MistSettings.TimeoutAttribute).asInstanceOf[Long])
continuation.removeAttribute(app.MistSettings.TimeoutAttribute)
None
}
@ -87,13 +87,13 @@ trait JettyContinuation extends ContinuationListener {
def suspended: Boolean = _continuation match {
case None ⇒ false
case Some(continuation) ⇒ (continuation.isSuspended || (continuation.getAttribute(application.MistSettings.TimeoutAttribute) ne null))
case Some(continuation) ⇒ (continuation.isSuspended || (continuation.getAttribute(app.MistSettings.TimeoutAttribute) ne null))
}
def timeout(ms: Long): Boolean = _continuation match {
case None ⇒ false
case Some(continuation) ⇒
continuation.setAttribute(application.MistSettings.TimeoutAttribute, ms)
continuation.setAttribute(app.MistSettings.TimeoutAttribute, ms)
continuation.resume
true
}
@ -103,21 +103,19 @@ trait JettyContinuation extends ContinuationListener {
//
def onComplete(c: Continuation) = {}
def onTimeout(c: Continuation) = {
c.getServletResponse.asInstanceOf[HttpServletResponse].addHeader(application.MistSettings.ExpiredHeaderName, application.MistSettings.ExpiredHeaderValue)
c.getServletResponse.asInstanceOf[HttpServletResponse].addHeader(app.MistSettings.ExpiredHeaderName, app.MistSettings.ExpiredHeaderValue)
c.complete
}
}
class JettyContinuationMethodFactory(val _application: AkkaApplication) extends RequestMethodFactory {
trait App {
def application = _application
}
def Delete(f: () ⇒ tAsyncRequestContext): RequestMethod = new Delete(f) with JettyContinuation with App
def Get(f: () ⇒ tAsyncRequestContext): RequestMethod = new Get(f) with JettyContinuation with App
def Head(f: () ⇒ tAsyncRequestContext): RequestMethod = new Head(f) with JettyContinuation with App
def Options(f: () ⇒ tAsyncRequestContext): RequestMethod = new Options(f) with JettyContinuation with App
def Post(f: () ⇒ tAsyncRequestContext): RequestMethod = new Post(f) with JettyContinuation with App
def Put(f: () ⇒ tAsyncRequestContext): RequestMethod = new Put(f) with JettyContinuation with App
def Trace(f: () ⇒ tAsyncRequestContext): RequestMethod = new Trace(f) with JettyContinuation with App
class JettyContinuationMethodFactory(_app: AkkaApplication) extends RequestMethodFactory {
implicit val app = _app
def Delete(f: () ⇒ tAsyncRequestContext): RequestMethod = new Delete(f) with JettyContinuation
def Get(f: () ⇒ tAsyncRequestContext): RequestMethod = new Get(f) with JettyContinuation
def Head(f: () ⇒ tAsyncRequestContext): RequestMethod = new Head(f) with JettyContinuation
def Options(f: () ⇒ tAsyncRequestContext): RequestMethod = new Options(f) with JettyContinuation
def Post(f: () ⇒ tAsyncRequestContext): RequestMethod = new Post(f) with JettyContinuation
def Put(f: () ⇒ tAsyncRequestContext): RequestMethod = new Put(f) with JettyContinuation
def Trace(f: () ⇒ tAsyncRequestContext): RequestMethod = new Trace(f) with JettyContinuation
}

40
akka-http/src/main/scala/akka/http/Mist.scala
View file

@ -53,7 +53,7 @@ object Types {
trait Mist {
import javax.servlet.ServletContext
protected def application: AkkaApplication
protected def app: AkkaApplication
/**
* The root endpoint actor
@ -99,7 +99,7 @@ trait Mist {
// shoot the message to the root endpoint for processing
// IMPORTANT: the suspend method is invoked on the server thread not in the actor
val method = builder(() ⇒ suspend(application.MistSettings.ConnectionClose))
val method = builder(() ⇒ suspend(app.MistSettings.ConnectionClose))
if (method.go) root ! method
}
@ -111,9 +111,9 @@ trait Mist {
val server = context.getServerInfo
val (major, minor) = (context.getMajorVersion, context.getMinorVersion)
factory = if (major >= 3) {
Some(new Servlet30ContextMethodFactory(application))
} else if (server.toLowerCase startsWith application.MistSettings.JettyServer) {
Some(new JettyContinuationMethodFactory(application))
Some(new Servlet30ContextMethodFactory(app))
} else if (server.toLowerCase startsWith app.MistSettings.JettyServer) {
Some(new JettyContinuationMethodFactory(app))
} else {
None
}
@ -123,14 +123,14 @@ trait Mist {
trait RootEndpointLocator {
var root: ActorRef = null
protected def application: AkkaApplication
protected def app: AkkaApplication
def configureRoot(address: String) {
def findRoot(address: String): ActorRef =
application.registry.actorFor(address).getOrElse(
app.registry.actorFor(address).getOrElse(
throw new ConfigurationException("akka.http.root-actor-id configuration option does not have a valid actor address [" + address + "]"))
root = if ((address eq null) || address == "") findRoot(application.MistSettings.RootActorID) else findRoot(address)
root = if ((address eq null) || address == "") findRoot(app.MistSettings.RootActorID) else findRoot(address)
}
}
@ -138,7 +138,7 @@ trait RootEndpointLocator {
* AkkaMistServlet adds support to bridge Http and Actors in an asynchronous fashion
* Async impls currently supported: Servlet3.0, Jetty Continuations
*/
class AkkaMistServlet(val application: AkkaApplication) extends HttpServlet with Mist with RootEndpointLocator {
class AkkaMistServlet(val app: AkkaApplication) extends HttpServlet with Mist with RootEndpointLocator {
import javax.servlet.{ ServletConfig }
/**
@ -157,7 +157,7 @@ class AkkaMistServlet(val application: AkkaApplication) extends HttpServlet with
* Proof-of-concept, use at own risk
* Will be officially supported in a later release
*/
class AkkaMistFilter(val application: AkkaApplication) extends Filter with Mist with RootEndpointLocator {
class AkkaMistFilter(val app: AkkaApplication) extends Filter with Mist with RootEndpointLocator {
import javax.servlet.{ ServletRequest, ServletResponse, FilterConfig, FilterChain }
/**
@ -294,6 +294,8 @@ class RootEndpoint extends Actor with Endpoint {
trait RequestMethod {
import java.io.IOException
import javax.servlet.http.{ HttpServletResponse, HttpServletRequest }
def app: AkkaApplication
// required implementations
val builder: () ⇒ tAsyncRequestContext
@ -358,7 +360,7 @@ trait RequestMethod {
}
} catch {
case io: Exception ⇒
EventHandler.error(io, this, io.getMessage)
app.eventHandler.error(io, this, io.getMessage)
false
}
case None ⇒ false
@ -374,7 +376,7 @@ trait RequestMethod {
}
} catch {
case io: IOException ⇒
EventHandler.error(io, this, io.getMessage)
app.eventHandler.error(io, this, io.getMessage)
}
case None ⇒ {}
}
@ -401,13 +403,13 @@ trait RequestMethod {
def Unavailable(body: String, retry: Int): Boolean = complete(HttpServletResponse.SC_SERVICE_UNAVAILABLE, body, List(("Retry-After", retry.toString)))
}
abstract class Delete(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Get(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Head(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Options(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Post(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Put(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Trace(val builder: () ⇒ tAsyncRequestContext) extends RequestMethod
abstract class Delete(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
abstract class Get(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
abstract class Head(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
abstract class Options(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
abstract class Post(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
abstract class Put(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
abstract class Trace(val builder: () ⇒ tAsyncRequestContext)(implicit val app: AkkaApplication) extends RequestMethod
trait RequestMethodFactory {
def Delete(f: () ⇒ tAsyncRequestContext): RequestMethod

31
akka-http/src/main/scala/akka/http/Servlet30Context.scala
View file

@ -6,7 +6,6 @@ package akka.http
import javax.servlet.{ AsyncContext, AsyncListener, AsyncEvent }
import Types._
import akka.event.EventHandler
import akka.AkkaApplication
/**
@ -15,7 +14,7 @@ import akka.AkkaApplication
trait Servlet30Context extends AsyncListener {
import javax.servlet.http.HttpServletResponse
protected def application: AkkaApplication
def app: AkkaApplication
val builder: () ⇒ tAsyncRequestContext
val context: Option[tAsyncRequestContext] = Some(builder())
@ -23,7 +22,7 @@ trait Servlet30Context extends AsyncListener {
protected val _ac: AsyncContext = {
val ac = context.get.asInstanceOf[AsyncContext]
ac setTimeout application.MistSettings.DefaultTimeout
ac setTimeout app.MistSettings.DefaultTimeout
ac addListener this
ac
}
@ -36,7 +35,7 @@ trait Servlet30Context extends AsyncListener {
true
} catch {
case e: IllegalStateException ⇒
EventHandler.error(e, this, e.getMessage)
app.eventHandler.error(e, this, e.getMessage)
false
}
}
@ -47,25 +46,23 @@ trait Servlet30Context extends AsyncListener {
def onComplete(e: AsyncEvent) {}
def onError(e: AsyncEvent) = e.getThrowable match {
case null ⇒
case t ⇒ EventHandler.error(t, this, t.getMessage)
case t ⇒ app.eventHandler.error(t, this, t.getMessage)
}
def onStartAsync(e: AsyncEvent) {}
def onTimeout(e: AsyncEvent) = {
e.getSuppliedResponse.asInstanceOf[HttpServletResponse].addHeader(application.MistSettings.ExpiredHeaderName, application.MistSettings.ExpiredHeaderValue)
e.getSuppliedResponse.asInstanceOf[HttpServletResponse].addHeader(app.MistSettings.ExpiredHeaderName, app.MistSettings.ExpiredHeaderValue)
e.getAsyncContext.complete
}
}
class Servlet30ContextMethodFactory(val _application: AkkaApplication) extends RequestMethodFactory {
trait App {
def application = _application
}
def Delete(f: () ⇒ tAsyncRequestContext): RequestMethod = new Delete(f) with Servlet30Context with App
def Get(f: () ⇒ tAsyncRequestContext): RequestMethod = new Get(f) with Servlet30Context with App
def Head(f: () ⇒ tAsyncRequestContext): RequestMethod = new Head(f) with Servlet30Context with App
def Options(f: () ⇒ tAsyncRequestContext): RequestMethod = new Options(f) with Servlet30Context with App
def Post(f: () ⇒ tAsyncRequestContext): RequestMethod = new Post(f) with Servlet30Context with App
def Put(f: () ⇒ tAsyncRequestContext): RequestMethod = new Put(f) with Servlet30Context with App
def Trace(f: () ⇒ tAsyncRequestContext): RequestMethod = new Trace(f) with Servlet30Context with App
class Servlet30ContextMethodFactory(_app: AkkaApplication) extends RequestMethodFactory {
implicit val app = _app
def Delete(f: () ⇒ tAsyncRequestContext): RequestMethod = new Delete(f) with Servlet30Context
def Get(f: () ⇒ tAsyncRequestContext): RequestMethod = new Get(f) with Servlet30Context
def Head(f: () ⇒ tAsyncRequestContext): RequestMethod = new Head(f) with Servlet30Context
def Options(f: () ⇒ tAsyncRequestContext): RequestMethod = new Options(f) with Servlet30Context
def Post(f: () ⇒ tAsyncRequestContext): RequestMethod = new Post(f) with Servlet30Context
def Put(f: () ⇒ tAsyncRequestContext): RequestMethod = new Put(f) with Servlet30Context
def Trace(f: () ⇒ tAsyncRequestContext): RequestMethod = new Trace(f) with Servlet30Context
}