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

Fixing ticket 808

Browse source
This commit is contained in:
Viktor Klang 2011-04-29 17:15:00 +02:00
parent b5873ff2c7
commit c2486cd52c
14 changed files with 95 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-actor/src/main/scala/akka/AkkaException.scala
View file

@ -16,7 +16,7 @@ import java.net.{InetAddress, UnknownHostException}
* *
* @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a> * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
*/ */
class AkkaException(message: String = "") extends RuntimeException(message) with Serializable { class AkkaException(message: String = "", cause: Throwable = null) extends RuntimeException(message, cause) with Serializable {
val uuid = "%s_%s".format(AkkaException.hostname, newUuid) val uuid = "%s_%s".format(AkkaException.hostname, newUuid)
override lazy val toString = { override lazy val toString = {

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

@ -67,12 +67,12 @@ case class MaximumNumberOfRestartsWithinTimeRangeReached(
@BeanProperty val lastExceptionCausingRestart: Throwable) extends LifeCycleMessage @BeanProperty val lastExceptionCausingRestart: Throwable) extends LifeCycleMessage
// Exceptions for Actors // Exceptions for Actors
class ActorStartException private[akka](message: String) extends AkkaException(message) class ActorStartException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
class IllegalActorStateException private[akka](message: String) extends AkkaException(message) class IllegalActorStateException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
class ActorKilledException private[akka](message: String) extends AkkaException(message) class ActorKilledException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
class ActorInitializationException private[akka](message: String) extends AkkaException(message) class ActorInitializationException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
class ActorTimeoutException private[akka](message: String) extends AkkaException(message) class ActorTimeoutException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
class InvalidMessageException private[akka](message: String) extends AkkaException(message) class InvalidMessageException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* This message is thrown by default when an Actors behavior doesn't match a message * This message is thrown by default when an Actors behavior doesn't match a message
@ -161,12 +161,15 @@ object Actor extends ListenerManagement {
*/ */
def actorOf(clazz: Class[_ <: Actor]): ActorRef = new LocalActorRef(() => { def actorOf(clazz: Class[_ <: Actor]): ActorRef = new LocalActorRef(() => {
import ReflectiveAccess.{ createInstance, noParams, noArgs } import ReflectiveAccess.{ createInstance, noParams, noArgs }
createInstance[Actor](clazz.asInstanceOf[Class[_]], noParams, noArgs).getOrElse( createInstance[Actor](clazz.asInstanceOf[Class[_]], noParams, noArgs) match {
throw new ActorInitializationException( case r: Right[Exception, Actor] => r.b
case l: Left[Exception, Actor] => throw new ActorInitializationException(
"Could not instantiate Actor of " + clazz + "Could not instantiate Actor of " + clazz +
"\nMake sure Actor is NOT defined inside a class/trait," + "\nMake sure Actor is NOT defined inside a class/trait," +
"\nif so put it outside the class/trait, f.e. in a companion object," + "\nif so put it outside the class/trait, f.e. in a companion object," +
"\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.")) "\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.", l.a)
}
}, None) }, None)
/** /**

2
akka-actor/src/main/scala/akka/actor/Supervisor.scala
View file

@ -13,7 +13,7 @@ import java.util.concurrent.{CopyOnWriteArrayList, ConcurrentHashMap}
import java.net.InetSocketAddress import java.net.InetSocketAddress
import akka.config.Supervision._ import akka.config.Supervision._
class SupervisorException private[akka](message: String) extends AkkaException(message) class SupervisorException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* Factory object for creating supervisors declarative. It creates instances of the 'Supervisor' class. * Factory object for creating supervisors declarative. It creates instances of the 'Supervisor' class.

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

@ -6,8 +6,8 @@ package akka.config
import akka.AkkaException import akka.AkkaException
class ConfigurationException(message: String) extends AkkaException(message) class ConfigurationException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
class ModuleNotAvailableException(message: String) extends AkkaException(message) class ModuleNotAvailableException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* Loads up the configuration (from the akka.conf file). * Loads up the configuration (from the akka.conf file).

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

@ -23,7 +23,7 @@ object DataFlow {
object Start object Start
object Exit object Exit
class DataFlowVariableException(msg: String) extends AkkaException(msg) class DataFlowVariableException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* Executes the supplied thunk in another thread. * Executes the supplied thunk in another thread.

16
akka-actor/src/main/scala/akka/dispatch/Dispatchers.scala
View file

@ -187,14 +187,14 @@ object Dispatchers {
case "GlobalExecutorBasedEventDriven" => GlobalExecutorBasedEventDrivenDispatcherConfigurator case "GlobalExecutorBasedEventDriven" => GlobalExecutorBasedEventDrivenDispatcherConfigurator
case fqn => case fqn =>
ReflectiveAccess.getClassFor[MessageDispatcherConfigurator](fqn) match { ReflectiveAccess.getClassFor[MessageDispatcherConfigurator](fqn) match {
case Some(clazz) => case r: Right[_, Class[MessageDispatcherConfigurator]] =>
val instance = ReflectiveAccess.createInstance[MessageDispatcherConfigurator](clazz, Array[Class[_]](), Array[AnyRef]()) ReflectiveAccess.createInstance[MessageDispatcherConfigurator](r.b, Array[Class[_]](), Array[AnyRef]()) match {
if (instance.isEmpty) case r: Right[Exception, MessageDispatcherConfigurator] => r.b
throw new IllegalArgumentException("Cannot instantiate MessageDispatcherConfigurator type [%s], make sure it has a default no-args constructor" format fqn) case l: Left[Exception, MessageDispatcherConfigurator] =>
else throw new IllegalArgumentException("Cannot instantiate MessageDispatcherConfigurator type [%s], make sure it has a default no-args constructor" format fqn, l.a)
instance.get }
case None => case l: Left[Exception, _] =>
throw new IllegalArgumentException("Unknown MessageDispatcherConfigurator type [%s]" format fqn) throw new IllegalArgumentException("Unknown MessageDispatcherConfigurator type [%s]" format fqn, l.a)
} }
} map { } map {
_ configure cfg _ configure cfg

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

@ -19,7 +19,7 @@ import java.util.{LinkedList => JLinkedList}
import scala.collection.mutable.Stack import scala.collection.mutable.Stack
import annotation.tailrec import annotation.tailrec
class FutureTimeoutException(message: String) extends AkkaException(message) class FutureTimeoutException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
object Futures { object Futures {

2
akka-actor/src/main/scala/akka/dispatch/MailboxHandling.scala
View file

@ -11,7 +11,7 @@ import java.util.{Queue, List, Comparator, PriorityQueue}
import java.util.concurrent._ import java.util.concurrent._
import akka.util._ import akka.util._
class MessageQueueAppendFailedException(message: String) extends AkkaException(message) class MessageQueueAppendFailedException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a> * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>

7
akka-actor/src/main/scala/akka/event/EventHandler.scala
View file

@ -220,14 +220,15 @@ object EventHandler extends ListenerManagement {
} }
defaultListeners foreach { listenerName => defaultListeners foreach { listenerName =>
try { try {
ReflectiveAccess.getClassFor[Actor](listenerName) map { clazz => ReflectiveAccess.getClassFor[Actor](listenerName) match {
addListener(Actor.actorOf(clazz).start()) case r: Right[_, Class[Actor]] => addListener(Actor.actorOf(r.b).start())
case l: Left[Exception,_] => throw l.a
} }
} catch { } catch {
case e: Exception => case e: Exception =>
throw new ConfigurationException( throw new ConfigurationException(
"Event Handler specified in config can't be loaded [" + listenerName + "Event Handler specified in config can't be loaded [" + listenerName +
"] due to [" + e.toString + "]") "] due to [" + e.toString + "]", e)
} }
} }
} }

32
akka-actor/src/main/scala/akka/remoteinterface/RemoteInterface.scala
View file

@ -116,7 +116,7 @@ case class RemoteServerWriteFailed(
class RemoteClientException private[akka] ( class RemoteClientException private[akka] (
message: String, message: String,
@BeanProperty val client: RemoteClientModule, @BeanProperty val client: RemoteClientModule,
val remoteAddress: InetSocketAddress) extends AkkaException(message) val remoteAddress: InetSocketAddress, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* Thrown when the remote server actor dispatching fails for some reason. * Thrown when the remote server actor dispatching fails for some reason.
@ -189,13 +189,14 @@ abstract class RemoteSupport extends ListenerManagement with RemoteServerModule
def actorOf(clazz: Class[_ <: Actor], host: String, port: Int): ActorRef = { def actorOf(clazz: Class[_ <: Actor], host: String, port: Int): ActorRef = {
import ReflectiveAccess.{ createInstance, noParams, noArgs } import ReflectiveAccess.{ createInstance, noParams, noArgs }
clientManagedActorOf(() => clientManagedActorOf(() =>
createInstance[Actor](clazz.asInstanceOf[Class[_]], noParams, noArgs).getOrElse( createInstance[Actor](clazz.asInstanceOf[Class[_]], noParams, noArgs) match {
throw new ActorInitializationException( case r: Right[_, Actor] => r.b
"Could not instantiate Actor" + case l: Left[Exception, _] => throw new ActorInitializationException(
"\nMake sure Actor is NOT defined inside a class/trait," + "Could not instantiate Actor" +
"\nif so put it outside the class/trait, f.e. in a companion object," + "\nMake sure Actor is NOT defined inside a class/trait," +
"\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.")), "\nif so put it outside the class/trait, f.e. in a companion object," +
host, port) "\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.", l.a)
}, host, port)
} }
/** /**
@ -217,13 +218,14 @@ abstract class RemoteSupport extends ListenerManagement with RemoteServerModule
def actorOf[T <: Actor : Manifest](host: String, port: Int): ActorRef = { def actorOf[T <: Actor : Manifest](host: String, port: Int): ActorRef = {
import ReflectiveAccess.{ createInstance, noParams, noArgs } import ReflectiveAccess.{ createInstance, noParams, noArgs }
clientManagedActorOf(() => clientManagedActorOf(() =>
createInstance[Actor](manifest[T].erasure.asInstanceOf[Class[_]], noParams, noArgs).getOrElse( createInstance[Actor](manifest[T].erasure.asInstanceOf[Class[_]], noParams, noArgs) match {
throw new ActorInitializationException( case r: Right[_, Actor] => r.b
"Could not instantiate Actor" + case l: Left[Exception, _] => throw new ActorInitializationException(
"\nMake sure Actor is NOT defined inside a class/trait," + "Could not instantiate Actor" +
"\nif so put it outside the class/trait, f.e. in a companion object," + "\nMake sure Actor is NOT defined inside a class/trait," +
"\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.")), "\nif so put it outside the class/trait, f.e. in a companion object," +
host, port) "\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.", l.a)
}, host, port)
} }
protected override def manageLifeCycleOfListeners = false protected override def manageLifeCycleOfListeners = false

72
akka-actor/src/main/scala/akka/util/ReflectiveAccess.scala
View file

@ -42,12 +42,16 @@ object ReflectiveAccess {
lazy val isEnabled = remoteSupportClass.isDefined lazy val isEnabled = remoteSupportClass.isDefined
def ensureEnabled = if (!isEnabled) { def ensureEnabled = if (!isEnabled) {
val e = new ModuleNotAvailableException( val e = new ModuleNotAvailableException("Can't load the remoting module, make sure that akka-remote.jar is on the classpath")
"Can't load the remoting module, make sure that akka-remote.jar is on the classpath")
EventHandler.debug(this, e.toString) EventHandler.debug(this, e.toString)
throw e throw e
} }
val remoteSupportClass: Option[Class[_ <: RemoteSupport]] = getClassFor(TRANSPORT) val remoteSupportClass = getClassFor[RemoteSupport](TRANSPORT) match {
case r: Right[_, Class[RemoteSupport]] => Some(r.b)
case l: Left[Exception,_] =>
EventHandler.debug(this, l.a)
None
}
protected[akka] val defaultRemoteSupport: Option[() => RemoteSupport] = protected[akka] val defaultRemoteSupport: Option[() => RemoteSupport] =
remoteSupportClass map { remoteClass => remoteSupportClass map { remoteClass =>
@ -55,9 +59,11 @@ object ReflectiveAccess {
remoteClass, remoteClass,
Array[Class[_]](), Array[Class[_]](),
Array[AnyRef]() Array[AnyRef]()
) getOrElse { ) match {
case r: Right[Exception, RemoteSupport] => r.b
case l: Left[Exception, RemoteSupport] =>
val e = new ModuleNotAvailableException( val e = new ModuleNotAvailableException(
"Can't instantiate [%s] - make sure that akka-remote.jar is on the classpath".format(remoteClass.getName)) "Can't instantiate [%s] - make sure that akka-remote.jar is on the classpath".format(remoteClass.getName), l.a)
EventHandler.debug(this, e.toString) EventHandler.debug(this, e.toString)
throw e throw e
} }
@ -114,7 +120,12 @@ object ReflectiveAccess {
getObjectFor("akka.cloud.cluster.Cluster$") getObjectFor("akka.cloud.cluster.Cluster$")
val serializerClass: Option[Class[_]] = val serializerClass: Option[Class[_]] =
getClassFor("akka.serialization.Serializer") getClassFor("akka.serialization.Serializer") match {
case r: Right[_, Class[_]] => Some(r.b)
case l: Left[Exception,_] =>
EventHandler.debug(this, l.toString)
None
}
def ensureEnabled = if (!isEnabled) throw new ModuleNotAvailableException( def ensureEnabled = if (!isEnabled) throw new ModuleNotAvailableException(
"Feature is only available in Akka Cloud") "Feature is only available in Akka Cloud")
@ -125,17 +136,15 @@ object ReflectiveAccess {
def createInstance[T](clazz: Class[_], def createInstance[T](clazz: Class[_],
params: Array[Class[_]], params: Array[Class[_]],
args: Array[AnyRef]): Option[T] = try { args: Array[AnyRef]): Either[Exception,T] = try {
assert(clazz ne null) assert(clazz ne null)
assert(params ne null) assert(params ne null)
assert(args ne null) assert(args ne null)
val ctor = clazz.getDeclaredConstructor(params: _*) val ctor = clazz.getDeclaredConstructor(params: _*)
ctor.setAccessible(true) ctor.setAccessible(true)
Some(ctor.newInstance(args: _*).asInstanceOf[T]) Right(ctor.newInstance(args: _*).asInstanceOf[T])
} catch { } catch {
case e: Exception => case e: Exception => Left(e)
EventHandler.debug(this, e.toString)
None
} }
def createInstance[T](fqn: String, def createInstance[T](fqn: String,
@ -145,11 +154,11 @@ object ReflectiveAccess {
assert(params ne null) assert(params ne null)
assert(args ne null) assert(args ne null)
getClassFor(fqn) match { getClassFor(fqn) match {
case Some(clazz) => case r: Right[Exception, Class[T]] =>
val ctor = clazz.getDeclaredConstructor(params: _*) val ctor = r.b.getDeclaredConstructor(params: _*)
ctor.setAccessible(true) ctor.setAccessible(true)
Some(ctor.newInstance(args: _*).asInstanceOf[T]) Some(ctor.newInstance(args: _*).asInstanceOf[T])
case None => None case _ => None
} }
} catch { } catch {
case e: Exception => case e: Exception =>
@ -159,11 +168,11 @@ object ReflectiveAccess {
def getObjectFor[T](fqn: String, classloader: ClassLoader = loader): Option[T] = try {//Obtains a reference to $MODULE$ def getObjectFor[T](fqn: String, classloader: ClassLoader = loader): Option[T] = try {//Obtains a reference to $MODULE$
getClassFor(fqn) match { getClassFor(fqn) match {
case Some(clazz) => case r: Right[Exception, Class[T]] =>
val instance = clazz.getDeclaredField("MODULE$") val instance = r.b.getDeclaredField("MODULE$")
instance.setAccessible(true) instance.setAccessible(true)
Option(instance.get(null).asInstanceOf[T]) Option(instance.get(null).asInstanceOf[T])
case None => None case _ => None
} }
} catch { } catch {
case e: ExceptionInInitializerError => case e: ExceptionInInitializerError =>
@ -171,45 +180,44 @@ object ReflectiveAccess {
throw e throw e
} }
def getClassFor[T](fqn: String, classloader: ClassLoader = loader): Option[Class[T]] = { def getClassFor[T](fqn: String, classloader: ClassLoader = loader): Either[Exception,Class[T]] = try {
assert(fqn ne null) assert(fqn ne null)
// First, use the specified CL // First, use the specified CL
val first = try { val first = try {
Option(classloader.loadClass(fqn).asInstanceOf[Class[T]]) Right(classloader.loadClass(fqn).asInstanceOf[Class[T]])
} catch { } catch {
case c: ClassNotFoundException => None case c: ClassNotFoundException => Left(c)
} }
if (first.isDefined) first if (first.isRight) first
else { else {
// Second option is to use the ContextClassLoader // Second option is to use the ContextClassLoader
val second = try { val second = try {
Option(Thread.currentThread.getContextClassLoader.loadClass(fqn).asInstanceOf[Class[T]]) Right(Thread.currentThread.getContextClassLoader.loadClass(fqn).asInstanceOf[Class[T]])
} catch { } catch {
case c: ClassNotFoundException => None case c: ClassNotFoundException => Left(c)
} }
if (second.isDefined) second if (second.isRight) second
else { else {
val third = try { val third = try {
// Don't try to use "loader" if we got the default "classloader" parameter if (classloader ne loader) Right(loader.loadClass(fqn).asInstanceOf[Class[T]]) else Left(null) //Horrid
if (classloader ne loader) Option(loader.loadClass(fqn).asInstanceOf[Class[T]])
else None
} catch { } catch {
case c: ClassNotFoundException => None case c: ClassNotFoundException => Left(c)
} }
if (third.isDefined) third if (third.isRight) third
else { else {
// Last option is Class.forName
try { try {
Option(Class.forName(fqn).asInstanceOf[Class[T]]) Right(Class.forName(fqn).asInstanceOf[Class[T]]) // Last option is Class.forName
} catch { } catch {
case c: ClassNotFoundException => None case c: ClassNotFoundException => Left(c)
} }
} }
} }
} }
} catch {
case e: Exception => Left(e)
} }
} }

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

@ -41,7 +41,7 @@ import java.util.concurrent.atomic.{AtomicReference, AtomicBoolean}
import java.util.concurrent._ import java.util.concurrent._
import akka.AkkaException import akka.AkkaException
class RemoteClientMessageBufferException(message: String) extends AkkaException(message) class RemoteClientMessageBufferException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
object RemoteEncoder { object RemoteEncoder {
def encode(rmp: RemoteMessageProtocol): AkkaRemoteProtocol = { def encode(rmp: RemoteMessageProtocol): AkkaRemoteProtocol = {

8
akka-testkit/src/main/scala/akka/testkit/TestActorRef.scala
View file

@ -72,12 +72,14 @@ object TestActorRef {
def apply[T <: Actor : Manifest] : TestActorRef[T] = new TestActorRef[T] ({ () => def apply[T <: Actor : Manifest] : TestActorRef[T] = new TestActorRef[T] ({ () =>
import ReflectiveAccess.{ createInstance, noParams, noArgs } import ReflectiveAccess.{ createInstance, noParams, noArgs }
createInstance[T](manifest[T].erasure, noParams, noArgs).getOrElse( createInstance[T](manifest[T].erasure, noParams, noArgs) match {
throw new ActorInitializationException( case r: Right[_, T] => r.b
case l: Left[Exception, _] => throw new ActorInitializationException(
"Could not instantiate Actor" + "Could not instantiate Actor" +
"\nMake sure Actor is NOT defined inside a class/trait," + "\nMake sure Actor is NOT defined inside a class/trait," +
"\nif so put it outside the class/trait, f.e. in a companion object," + "\nif so put it outside the class/trait, f.e. in a companion object," +
"\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.")) "\nOR try to change: 'actorOf[MyActor]' to 'actorOf(new MyActor)'.", l.a)
}
}) })
} }

2
akka-typed-actor/src/main/scala/akka/transactor/Coordination.scala
View file

@ -9,7 +9,7 @@ import akka.stm.Atomic
import scala.util.DynamicVariable import scala.util.DynamicVariable
class CoordinateException private[akka](message: String) extends AkkaException(message) class CoordinateException private[akka](message: String, cause: Throwable = null) extends AkkaException(message, cause)
/** /**
* Coordinating transactions between typed actors. * Coordinating transactions between typed actors.