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

Merge with master

Browse source
This commit is contained in:
Viktor Klang 2011-03-14 16:29:22 +01:00
commit 0899fa44ab
15 changed files with 83 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

57
akka-actor/src/main/scala/akka/actor/Actor.scala
View file

@ -97,7 +97,11 @@ class ActorTimeoutException private[akka](message: String) extends AkkaEx
*
* Log an error event:
* <pre>
* EventHandler.notifyListeners(EventHandler.Error(exception, this, message.toString))
* EventHandler.notify(EventHandler.Error(exception, this, message.toString))
* </pre>
* Or use the direct methods (better performance):
* <pre>
* EventHandler.error(exception, this, message.toString)
* </pre>
* @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
*/
@ -114,15 +118,47 @@ object EventHandler extends ListenerManagement {
case class Warning(instance: AnyRef, message: String = "") extends Event
case class Info(instance: AnyRef, message: String = "") extends Event
case class Debug(instance: AnyRef, message: String = "") extends Event
val error = "[ERROR] [%s] [%s] [%s] %s\n%s".intern
val warning = "[WARN] [%s] [%s] [%s] %s".intern
val info = "[INFO] [%s] [%s] [%s] %s".intern
val debug = "[DEBUG] [%s] [%s] [%s] %s".intern
val ID = "default:error:handler".intern
val error = "[ERROR] [%s] [%s] [%s] %s\n%s".intern
val warning = "[WARN] [%s] [%s] [%s] %s".intern
val info = "[INFO] [%s] [%s] [%s] %s".intern
val debug = "[DEBUG] [%s] [%s] [%s] %s".intern
val generic = "[GENERIC] [%s] [%s]".intern
val ID = "event:handler".intern
val EventHandlerDispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher(ID).build
val level: Class[_ <: Event] = config.getString("akka.event-handler-level", "DEBUG") match {
case "ERROR" => classOf[Error]
case "WARNING" => classOf[Warning]
case "INFO" => classOf[Info]
case "DEBUG" => classOf[Debug]
case unknown => throw new ConfigurationException(
"Configuration option 'akka.event-handler-level' is invalid [" + unknown + "]")
}
def notify(event: => AnyRef) = notifyListeners(event)
def notify[T <: Event : ClassManifest](event: => T) {
if (classManifest[T].erasure.asInstanceOf[Class[_ <: Event]] == level) notifyListeners(event)
}
def error(cause: Throwable, instance: AnyRef, message: => String) = {
if (level == classOf[Error]) notifyListeners(Error(cause, instance, message))
}
def warning(instance: AnyRef, message: => String) = {
if (level == classOf[Warning]) notifyListeners(Warning(instance, message))
}
def info(instance: AnyRef, message: => String) = {
if (level == classOf[Info]) notifyListeners(Info(instance, message))
}
def debug(instance: AnyRef, message: => String) = {
if (level == classOf[Debug]) notifyListeners(Debug(instance, message))
}
def formattedTimestamp = DateFormat.getInstance.format(new Date)
def stackTraceFor(e: Throwable) = {
@ -162,7 +198,8 @@ object EventHandler extends ListenerManagement {
event.thread.getName,
instance.getClass.getSimpleName,
message))
case _ => {}
case event =>
println(generic.format(formattedTimestamp, event.toString))
}
}
@ -232,9 +269,7 @@ object Actor extends ListenerManagement {
type Receive = PartialFunction[Any, Unit]
private[actor] val actorRefInCreation = new scala.util.DynamicVariable[Option[ActorRef]](None)
/**
* Creates an ActorRef out of the Actor with type T.
* <pre>

4
akka-actor/src/main/scala/akka/actor/ActorRef.scala
View file

@ -824,7 +824,7 @@ class LocalActorRef private[akka] (
}
} catch {
case e: Throwable =>
EventHandler notifyListeners EventHandler.Error(e, this, messageHandle.message.toString)
EventHandler notify EventHandler.Error(e, this, messageHandle.message.toString)
throw e
}
}
@ -987,7 +987,7 @@ class LocalActorRef private[akka] (
}
private def handleExceptionInDispatch(reason: Throwable, message: Any) = {
EventHandler notifyListeners EventHandler.Error(reason, this, message.toString)
EventHandler notify EventHandler.Error(reason, this, message.toString)
//Prevent any further messages to be processed until the actor has been restarted
dispatcher.suspend(this)

8
akka-actor/src/main/scala/akka/actor/Scheduler.scala
View file

@ -38,7 +38,7 @@ object Scheduler {
initialDelay, delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this, receiver + " @ " + message)
EventHandler notify EventHandler.Error(e, this, receiver + " @ " + message)
throw SchedulerException(message + " could not be scheduled on " + receiver, e)
}
}
@ -59,7 +59,7 @@ object Scheduler {
service.scheduleAtFixedRate(runnable, initialDelay, delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
throw SchedulerException("Failed to schedule a Runnable", e)
}
}
@ -74,7 +74,7 @@ object Scheduler {
delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this, receiver + " @ " + message)
EventHandler notify EventHandler.Error(e, this, receiver + " @ " + message)
throw SchedulerException( message + " could not be scheduleOnce'd on " + receiver, e)
}
}
@ -95,7 +95,7 @@ object Scheduler {
service.schedule(runnable,delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
throw SchedulerException("Failed to scheduleOnce a Runnable", e)
}
}

2
akka-actor/src/main/scala/akka/config/Config.scala
View file

@ -77,7 +77,7 @@ object Config {
}
} catch {
case e =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
throw e
}
}

2
akka-actor/src/main/scala/akka/dataflow/DataFlow.scala
View file

@ -148,7 +148,7 @@ object DataFlow {
(out !! Get).as[T]
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
out ! Exit
throw e
}

6
akka-actor/src/main/scala/akka/dispatch/ExecutorBasedEventDrivenDispatcher.scala
View file

@ -102,7 +102,7 @@ class ExecutorBasedEventDrivenDispatcher(
try executorService.get() execute invocation
catch {
case e: RejectedExecutionException =>
EventHandler notifyListeners EventHandler.Warning(this, e.toString)
EventHandler notify EventHandler.Warning(this, e.toString)
throw e
}
}
@ -149,7 +149,7 @@ class ExecutorBasedEventDrivenDispatcher(
executorService.get() execute mbox
} catch {
case e: RejectedExecutionException =>
EventHandler notifyListeners EventHandler.Warning(this, e.toString)
EventHandler notify EventHandler.Warning(this, e.toString)
mbox.dispatcherLock.unlock()
throw e
}
@ -273,4 +273,4 @@ trait PriorityMailbox { self: ExecutorBasedEventDrivenDispatcher =>
final def dispatcher = self
}
}
}
}

8
akka-actor/src/main/scala/akka/dispatch/Future.scala
View file

@ -90,7 +90,7 @@ object Futures {
result completeWithResult r
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
result completeWithException e
}
}
@ -289,7 +289,7 @@ sealed trait Future[+T] {
Right(f(v.right.get))
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
Left(e)
})
}
@ -317,7 +317,7 @@ sealed trait Future[+T] {
f(v.right.get) onComplete (fa.completeWith(_))
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
fa completeWithException e
}
}
@ -347,7 +347,7 @@ sealed trait Future[+T] {
else Left(new MatchError(r))
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
Left(e)
})
}

8
akka-actor/src/main/scala/akka/dispatch/MessageHandling.scala
View file

@ -16,9 +16,9 @@ import akka.actor._
* @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
*/
final case class MessageInvocation(val receiver: ActorRef,
val message: Any,
val sender: Option[ActorRef],
val senderFuture: Option[CompletableFuture[Any]]) {
val message: Any,
val sender: Option[ActorRef],
val senderFuture: Option[CompletableFuture[Any]]) {
if (receiver eq null) throw new IllegalArgumentException("Receiver can't be null")
def invoke = try {
@ -36,7 +36,7 @@ final case class FutureInvocation(future: CompletableFuture[Any], function: () =
Right(function.apply)
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
Left(e)
})
}

4
akka-actor/src/main/scala/akka/dispatch/ThreadPoolBuilder.scala
View file

@ -208,10 +208,10 @@ class BoundedExecutorDecorator(val executor: ExecutorService, bound: Int) extend
})
} catch {
case e: RejectedExecutionException =>
EventHandler notifyListeners EventHandler.Warning(this, e.toString)
EventHandler notify EventHandler.Warning(this, e.toString)
semaphore.release
case e: Throwable =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
throw e
}
}

2
akka-actor/src/main/scala/akka/util/LockUtil.scala
View file

@ -126,7 +126,7 @@ class Switch(startAsOn: Boolean = false) {
action
} catch {
case t: Throwable =>
EventHandler notifyListeners EventHandler.Error(t, this)
EventHandler notify EventHandler.Error(t, this)
switch.compareAndSet(!from, from) //Revert status
throw t
}

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

@ -389,7 +389,7 @@ trait RequestMethod {
}
} catch {
case io: Exception =>
EventHandler notifyListeners EventHandler.Error(io, this)
EventHandler notify EventHandler.Error(io, this)
false
}
}
@ -408,7 +408,7 @@ trait RequestMethod {
}
} catch {
case io: IOException =>
EventHandler notifyListeners EventHandler.Error(io, this)
EventHandler notify EventHandler.Error(io, this)
}
}

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

@ -36,7 +36,7 @@ trait Servlet30Context extends AsyncListener {
}
catch {
case ex: IllegalStateException =>
EventHandler notifyListeners EventHandler.Error(ex, this)
EventHandler notify EventHandler.Error(ex, this)
false
}
}

2
akka-http/src/main/scala/akka/security/Security.scala
View file

@ -369,7 +369,7 @@ trait SpnegoAuthenticationActor extends AuthenticationActor[SpnegoCredentials] {
Some(UserInfo(user, null, rolesFor(user)))
} catch {
case e: PrivilegedActionException => {
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
None
}
}

20
akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala
View file

@ -429,7 +429,7 @@ class ActiveRemoteClientHandler(
}
} catch {
case e: Throwable =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
client.notifyListeners(RemoteClientError(e, client.module, client.remoteAddress))
throw e
}
@ -478,7 +478,7 @@ class ActiveRemoteClientHandler(
.newInstance(exception.getMessage).asInstanceOf[Throwable]
} catch {
case problem: Throwable =>
EventHandler notifyListeners EventHandler.Error(problem, this)
EventHandler notify EventHandler.Error(problem, this)
UnparsableException(classname, exception.getMessage)
}
}
@ -558,7 +558,7 @@ class NettyRemoteServer(serverModule: NettyRemoteServerModule, val host: String,
serverModule.notifyListeners(RemoteServerShutdown(serverModule))
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
}
}
}
@ -591,7 +591,7 @@ trait NettyRemoteServerModule extends RemoteServerModule { self: RemoteModule =>
}
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
notifyListeners(RemoteServerError(e, this))
}
this
@ -862,7 +862,7 @@ class RemoteServerHandler(
val actorRef =
try { createActor(actorInfo, channel).start } catch {
case e: SecurityException =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
write(channel, createErrorReplyMessage(e, request, AkkaActorType.ScalaActor))
server.notifyListeners(RemoteServerError(e, server))
return
@ -981,7 +981,7 @@ class RemoteServerHandler(
write(channel, RemoteEncoder.encode(messageBuilder.build))
} catch {
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
server.notifyListeners(RemoteServerError(e, server))
}
@ -997,11 +997,11 @@ class RemoteServerHandler(
}
} catch {
case e: InvocationTargetException =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
write(channel, createErrorReplyMessage(e.getCause, request, AkkaActorType.TypedActor))
server.notifyListeners(RemoteServerError(e, server))
case e: Exception =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
write(channel, createErrorReplyMessage(e, request, AkkaActorType.TypedActor))
server.notifyListeners(RemoteServerError(e, server))
}
@ -1061,7 +1061,7 @@ class RemoteServerHandler(
actorRef
} catch {
case e: Throwable =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
server.notifyListeners(RemoteServerError(e, server))
throw e
}
@ -1128,7 +1128,7 @@ class RemoteServerHandler(
newInstance
} catch {
case e: Throwable =>
EventHandler notifyListeners EventHandler.Error(e, this)
EventHandler notify EventHandler.Error(e, this)
server.notifyListeners(RemoteServerError(e, server))
throw e
}

1
config/akka-reference.conf
View file

@ -13,6 +13,7 @@ akka {
time-unit = "seconds" # Time unit for all timeout properties throughout the config
event-handlers = ["akka.actor.EventHandler$DefaultListener"] # event handlers to register at boot time (EventHandler$DefaultListener logs to STDOUT)
event-handler-level = "DEBUG" # Options: ERROR, WARNING, INFO, DEBUG
# These boot classes are loaded (and created) automatically when the Akka Microkernel boots up
# Can be used to bootstrap your application(s)