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

Fixed bug in ActorRegistry getting typed actor by manifest

Browse source
This commit is contained in:
Jonas Bonér 2010-11-22 08:59:45 +01:00
parent 9c072327d9
commit 4f8b3ea6fb
6 changed files with 106 additions and 106 deletions
Download patch file Download diff file Expand all files Collapse all files

101
akka-actor/src/main/scala/akka/actor/ActorRegistry.scala
View file

@ -37,10 +37,10 @@ case class ActorUnregistered(actor: ActorRef) extends ActorRegistryEvent
* @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
*/
object ActorRegistry extends ListenerManagement {
private val actorsByUUID = new ConcurrentHashMap[Uuid, ActorRef]
private val actorsById = new Index[String,ActorRef]
private val remoteActorSets = Map[Address, RemoteActorSet]()
private val guard = new ReadWriteGuard
private val actorsByUUID = new ConcurrentHashMap[Uuid, ActorRef]
private val actorsById = new Index[String,ActorRef]
private val remoteActorSets = Map[Address, RemoteActorSet]()
private val guard = new ReadWriteGuard
/**
* Returns all actors in the system.
@ -50,7 +50,7 @@ object ActorRegistry extends ListenerManagement {
/**
* Returns the number of actors in the system.
*/
def size : Int = actorsByUUID.size
def size: Int = actorsByUUID.size
/**
* Invokes a function for all actors.
@ -68,8 +68,7 @@ object ActorRegistry extends ListenerManagement {
val elements = actorsByUUID.elements
while (elements.hasMoreElements) {
val element = elements.nextElement
if(f isDefinedAt element)
return Some(f(element))
if (f isDefinedAt element) return Some(f(element))
}
None
}
@ -88,9 +87,7 @@ object ActorRegistry extends ListenerManagement {
val elements = actorsByUUID.elements
while (elements.hasMoreElements) {
val actorId = elements.nextElement
if (p(actorId)) {
all += actorId
}
if (p(actorId)) all += actorId
}
all.toArray
}
@ -105,7 +102,7 @@ object ActorRegistry extends ListenerManagement {
* Finds any actor that matches T.
*/
def actorFor[T <: Actor](implicit manifest: Manifest[T]): Option[ActorRef] =
find({ case a:ActorRef if manifest.erasure.isAssignableFrom(a.actor.getClass) => a })
find({ case a: ActorRef if manifest.erasure.isAssignableFrom(a.actor.getClass) => a })
/**
* Finds all actors of type or sub-type specified by the class passed in as the Class argument.
@ -132,13 +129,11 @@ object ActorRegistry extends ListenerManagement {
* Invokes a function for all typed actors.
*/
def foreachTypedActor(f: (AnyRef) => Unit) = {
TypedActorModule.ensureTypedActorEnabled
TypedActorModule.ensureEnabled
val elements = actorsByUUID.elements
while (elements.hasMoreElements) {
val proxy = typedActorFor(elements.nextElement)
if (proxy.isDefined) {
f(proxy.get)
}
if (proxy.isDefined) f(proxy.get)
}
}
@ -147,12 +142,11 @@ object ActorRegistry extends ListenerManagement {
* Returns None if the function never returns Some
*/
def findTypedActor[T](f: PartialFunction[AnyRef,T]) : Option[T] = {
TypedActorModule.ensureTypedActorEnabled
TypedActorModule.ensureEnabled
val elements = actorsByUUID.elements
while (elements.hasMoreElements) {
val proxy = typedActorFor(elements.nextElement)
if(proxy.isDefined && (f isDefinedAt proxy))
return Some(f(proxy))
if (proxy.isDefined && (f isDefinedAt proxy)) return Some(f(proxy))
}
None
}
@ -161,14 +155,12 @@ object ActorRegistry extends ListenerManagement {
* Finds all typed actors that satisfy a predicate.
*/
def filterTypedActors(p: AnyRef => Boolean): Array[AnyRef] = {
TypedActorModule.ensureTypedActorEnabled
TypedActorModule.ensureEnabled
val all = new ListBuffer[AnyRef]
val elements = actorsByUUID.elements
while (elements.hasMoreElements) {
val proxy = typedActorFor(elements.nextElement)
if (proxy.isDefined && p(proxy.get)) {
all += proxy.get
}
if (proxy.isDefined && p(proxy.get)) all += proxy.get
}
all.toArray
}
@ -177,7 +169,7 @@ object ActorRegistry extends ListenerManagement {
* Finds all typed actors that are subtypes of the class passed in as the Manifest argument.
*/
def typedActorsFor[T <: AnyRef](implicit manifest: Manifest[T]): Array[AnyRef] = {
TypedActorModule.ensureTypedActorEnabled
TypedActorModule.ensureEnabled
typedActorsFor[T](manifest.erasure.asInstanceOf[Class[T]])
}
@ -185,20 +177,20 @@ object ActorRegistry extends ListenerManagement {
* Finds any typed actor that matches T.
*/
def typedActorFor[T <: AnyRef](implicit manifest: Manifest[T]): Option[AnyRef] = {
TypedActorModule.ensureTypedActorEnabled
def predicate(proxy: AnyRef) : Boolean = {
TypedActorModule.ensureEnabled
def predicate(proxy: AnyRef): Boolean = {
val actorRef = TypedActorModule.typedActorObjectInstance.get.actorFor(proxy)
actorRef.isDefined && manifest.erasure.isAssignableFrom(actorRef.get.actor.getClass)
}
findTypedActor({ case a:AnyRef if predicate(a) => a })
findTypedActor({ case a: Some[AnyRef] if predicate(a.get) => a })
}
/**
* Finds all typed actors of type or sub-type specified by the class passed in as the Class argument.
*/
def typedActorsFor[T <: AnyRef](clazz: Class[T]): Array[AnyRef] = {
TypedActorModule.ensureTypedActorEnabled
def predicate(proxy: AnyRef) : Boolean = {
TypedActorModule.ensureEnabled
def predicate(proxy: AnyRef): Boolean = {
val actorRef = TypedActorModule.typedActorObjectInstance.get.actorFor(proxy)
actorRef.isDefined && clazz.isAssignableFrom(actorRef.get.actor.getClass)
}
@ -209,7 +201,7 @@ object ActorRegistry extends ListenerManagement {
* Finds all typed actors that have a specific id.
*/
def typedActorsFor(id: String): Array[AnyRef] = {
TypedActorModule.ensureTypedActorEnabled
TypedActorModule.ensureEnabled
val actorRefs = actorsById values id
actorRefs.flatMap(typedActorFor(_))
}
@ -218,12 +210,10 @@ object ActorRegistry extends ListenerManagement {
* Finds the typed actor that has a specific UUID.
*/
def typedActorFor(uuid: Uuid): Option[AnyRef] = {
TypedActorModule.ensureTypedActorEnabled
TypedActorModule.ensureEnabled
val actorRef = actorsByUUID get uuid
if (actorRef eq null)
None
else
typedActorFor(actorRef)
if (actorRef eq null) None
else typedActorFor(actorRef)
}
/**
@ -265,20 +255,15 @@ object ActorRegistry extends ListenerManagement {
*/
def shutdownAll() {
log.info("Shutting down all actors in the system...")
if (TypedActorModule.isTypedActorEnabled) {
if (TypedActorModule.isEnabled) {
val elements = actorsByUUID.elements
while (elements.hasMoreElements) {
val actorRef = elements.nextElement
val proxy = typedActorFor(actorRef)
if (proxy.isDefined) {
TypedActorModule.typedActorObjectInstance.get.stop(proxy.get)
} else {
actorRef.stop
}
if (proxy.isDefined) TypedActorModule.typedActorObjectInstance.get.stop(proxy.get)
else actorRef.stop
}
} else {
foreach(_.stop)
}
} else foreach(_.stop)
actorsByUUID.clear
actorsById.clear
log.info("All actors have been shut down and unregistered from ActorRegistry")
@ -337,16 +322,13 @@ class Index[K <: AnyRef,V <: AnyRef : Manifest] {
if (set ne null) {
set.synchronized {
if (set.isEmpty) {
retry = true //IF the set is empty then it has been removed, so signal retry
}
if (set.isEmpty) retry = true //IF the set is empty then it has been removed, so signal retry
else { //Else add the value to the set and signal that retry is not needed
added = set add v
retry = false
}
}
}
else {
} else {
val newSet = new ConcurrentSkipListSet[V]
newSet add v
@ -354,24 +336,20 @@ class Index[K <: AnyRef,V <: AnyRef : Manifest] {
val oldSet = container.putIfAbsent(k,newSet)
if (oldSet ne null) {
oldSet.synchronized {
if (oldSet.isEmpty) {
retry = true //IF the set is empty then it has been removed, so signal retry
}
if (oldSet.isEmpty) retry = true //IF the set is empty then it has been removed, so signal retry
else { //Else try to add the value to the set and signal that retry is not needed
added = oldSet add v
retry = false
}
}
} else {
added = true
}
} else added = true
}
if (retry) spinPut(k,v)
if (retry) spinPut(k, v)
else added
}
spinPut(key,value)
spinPut(key, value)
}
/**
@ -390,10 +368,8 @@ class Index[K <: AnyRef,V <: AnyRef : Manifest] {
def findValue(key: K)(f: (V) => Boolean): Option[V] = {
import scala.collection.JavaConversions._
val set = container get key
if (set ne null)
set.iterator.find(f)
else
None
if (set ne null) set.iterator.find(f)
else None
}
/**
@ -420,8 +396,7 @@ class Index[K <: AnyRef,V <: AnyRef : Manifest] {
container.remove(key,emptySet) //We try to remove the key if it's mapped to an empty set
true //Remove succeeded
}
else false //Remove failed
} else false //Remove failed
}
} else false //Remove failed
}
@ -434,5 +409,5 @@ class Index[K <: AnyRef,V <: AnyRef : Manifest] {
/**
* Removes all keys and all values
*/
def clear = foreach { case (k,v) => remove(k,v) }
def clear = foreach { case (k, v) => remove(k, v) }
}

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

@ -5,7 +5,7 @@
package akka.dispatch
import akka.actor.{ActorRef, IllegalActorStateException}
import akka.util.ReflectiveAccess.EnterpriseModule
import akka.util.ReflectiveAccess.AkkaCloudModule
import java.util.Queue
import akka.util.Switch
@ -123,10 +123,10 @@ class ExecutorBasedEventDrivenDispatcher(
*/
def createDurableMailbox(actorRef: ActorRef, mailboxType: DurableMailboxType): AnyRef = mailboxType match {
// FIXME make generic (work for TypedActor as well)
case FileBasedDurableMailbox(serializer) => EnterpriseModule.createFileBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case ZooKeeperBasedDurableMailbox(serializer) => EnterpriseModule.createZooKeeperBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case BeanstalkBasedDurableMailbox(serializer) => EnterpriseModule.createBeanstalkBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case RedisBasedDurableMailbox(serializer) => EnterpriseModule.createRedisBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case FileBasedDurableMailbox(serializer) => AkkaCloudModule.createFileBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case ZooKeeperBasedDurableMailbox(serializer) => AkkaCloudModule.createZooKeeperBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case BeanstalkBasedDurableMailbox(serializer) => AkkaCloudModule.createBeanstalkBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case RedisBasedDurableMailbox(serializer) => AkkaCloudModule.createRedisBasedMailbox(actorRef).asInstanceOf[MessageQueue]
case AMQPBasedDurableMailbox(serializer) => throw new UnsupportedOperationException("AMQPBasedDurableMailbox is not yet supported")
case JMSBasedDurableMailbox(serializer) => throw new UnsupportedOperationException("JMSBasedDurableMailbox is not yet supported")
}

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

@ -5,7 +5,7 @@
package akka.dispatch
import akka.actor.{Actor, ActorType, ActorRef, ActorInitializationException}
import akka.util.ReflectiveAccess.EnterpriseModule
import akka.util.ReflectiveAccess.AkkaCloudModule
import akka.AkkaException
import java.util.{Queue, List}
@ -42,15 +42,15 @@ case class BoundedMailbox(
if (pushTimeOut eq null) throw new IllegalArgumentException("The push time-out for BoundedMailbox can not be null")
}
abstract class DurableMailboxType(val serializer: EnterpriseModule.Serializer) extends MailboxType {
abstract class DurableMailboxType(val serializer: AkkaCloudModule.Serializer) extends MailboxType {
if (serializer eq null) throw new IllegalArgumentException("The serializer for DurableMailboxType can not be null")
}
case class FileBasedDurableMailbox(ser: EnterpriseModule.Serializer) extends DurableMailboxType(ser)
case class RedisBasedDurableMailbox(ser: EnterpriseModule.Serializer) extends DurableMailboxType(ser)
case class BeanstalkBasedDurableMailbox(ser: EnterpriseModule.Serializer) extends DurableMailboxType(ser)
case class ZooKeeperBasedDurableMailbox(ser: EnterpriseModule.Serializer) extends DurableMailboxType(ser)
case class AMQPBasedDurableMailbox(ser: EnterpriseModule.Serializer) extends DurableMailboxType(ser)
case class JMSBasedDurableMailbox(ser: EnterpriseModule.Serializer) extends DurableMailboxType(ser)
case class FileBasedDurableMailbox(ser: AkkaCloudModule.Serializer) extends DurableMailboxType(ser)
case class RedisBasedDurableMailbox(ser: AkkaCloudModule.Serializer) extends DurableMailboxType(ser)
case class BeanstalkBasedDurableMailbox(ser: AkkaCloudModule.Serializer) extends DurableMailboxType(ser)
case class ZooKeeperBasedDurableMailbox(ser: AkkaCloudModule.Serializer) extends DurableMailboxType(ser)
case class AMQPBasedDurableMailbox(ser: AkkaCloudModule.Serializer) extends DurableMailboxType(ser)
case class JMSBasedDurableMailbox(ser: AkkaCloudModule.Serializer) extends DurableMailboxType(ser)
class DefaultUnboundedMessageQueue(blockDequeue: Boolean)
extends LinkedBlockingQueue[MessageInvocation] with MessageQueue {

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

@ -20,13 +20,13 @@ object ReflectiveAccess extends Logging {
val loader = getClass.getClassLoader
lazy val isRemotingEnabled = RemoteClientModule.isRemotingEnabled
lazy val isTypedActorEnabled = TypedActorModule.isTypedActorEnabled
lazy val isEnterpriseEnabled = EnterpriseModule.isEnterpriseEnabled
lazy val isRemotingEnabled = RemoteClientModule.isEnabled
lazy val isTypedActorEnabled = TypedActorModule.isEnabled
lazy val isAkkaCloudEnabled = AkkaCloudModule.isEnabled
def ensureRemotingEnabled = RemoteClientModule.ensureRemotingEnabled
def ensureTypedActorEnabled = TypedActorModule.ensureTypedActorEnabled
def ensureEnterpriseEnabled = EnterpriseModule.ensureEnterpriseEnabled
def ensureRemotingEnabled = RemoteClientModule.ensureEnabled
def ensureTypedActorEnabled = TypedActorModule.ensureEnabled
def ensureAkkaCloudEnabled = AkkaCloudModule.ensureEnabled
/**
* Reflective access to the RemoteClient module.
@ -56,32 +56,32 @@ object ReflectiveAccess extends Logging {
def clientFor(hostname: String, port: Int, loader: Option[ClassLoader]): RemoteClient
}
lazy val isRemotingEnabled = remoteClientObjectInstance.isDefined
lazy val isEnabled = remoteClientObjectInstance.isDefined
def ensureRemotingEnabled = if (!isRemotingEnabled) throw new ModuleNotAvailableException(
def ensureEnabled = if (!isEnabled) throw new ModuleNotAvailableException(
"Can't load the remoting module, make sure that akka-remote.jar is on the classpath")
val remoteClientObjectInstance: Option[RemoteClientObject] =
getObjectFor("akka.remote.RemoteClient$")
def register(address: InetSocketAddress, uuid: Uuid) = {
ensureRemotingEnabled
ensureEnabled
remoteClientObjectInstance.get.register(address.getHostName, address.getPort, uuid)
}
def unregister(address: InetSocketAddress, uuid: Uuid) = {
ensureRemotingEnabled
ensureEnabled
remoteClientObjectInstance.get.unregister(address.getHostName, address.getPort, uuid)
}
def registerSupervisorForActor(remoteAddress: InetSocketAddress, actorRef: ActorRef) = {
ensureRemotingEnabled
ensureEnabled
val remoteClient = remoteClientObjectInstance.get.clientFor(remoteAddress)
remoteClient.registerSupervisorForActor(actorRef)
}
def clientFor(hostname: String, port: Int, loader: Option[ClassLoader]): RemoteClient = {
ensureRemotingEnabled
ensureEnabled
remoteClientObjectInstance.get.clientFor(hostname, port, loader)
}
@ -95,7 +95,7 @@ object ReflectiveAccess extends Logging {
actorRef: ActorRef,
typedActorInfo: Option[Tuple2[String, String]],
actorType: ActorType): Option[CompletableFuture[T]] = {
ensureRemotingEnabled
ensureEnabled
clientFor(remoteAddress.getHostName, remoteAddress.getPort, None).send[T](
message, senderOption, senderFuture, remoteAddress, timeout, isOneWay, actorRef, typedActorInfo, actorType)
}
@ -126,17 +126,17 @@ object ReflectiveAccess extends Logging {
getObjectFor("akka.remote.RemoteNode$")
def registerActor(address: InetSocketAddress, actorRef: ActorRef) = {
ensureRemotingEnabled
RemoteClientModule.ensureEnabled
remoteServerObjectInstance.get.registerActor(address, actorRef)
}
def registerTypedActor(address: InetSocketAddress, implementationClassName: String, proxy: AnyRef) = {
ensureRemotingEnabled
RemoteClientModule.ensureEnabled
remoteServerObjectInstance.get.registerTypedActor(address, implementationClassName, proxy)
}
def unregister(actorRef: ActorRef) = {
ensureRemotingEnabled
RemoteClientModule.ensureEnabled
remoteNodeObjectInstance.get.unregister(actorRef)
}
}
@ -156,16 +156,16 @@ object ReflectiveAccess extends Logging {
def stop(anyRef: AnyRef) : Unit
}
lazy val isTypedActorEnabled = typedActorObjectInstance.isDefined
lazy val isEnabled = typedActorObjectInstance.isDefined
def ensureTypedActorEnabled = if (!isTypedActorEnabled) throw new ModuleNotAvailableException(
def ensureEnabled = if (!isTypedActorEnabled) throw new ModuleNotAvailableException(
"Can't load the typed actor module, make sure that akka-typed-actor.jar is on the classpath")
val typedActorObjectInstance: Option[TypedActorObject] =
getObjectFor("akka.actor.TypedActor$")
def resolveFutureIfMessageIsJoinPoint(message: Any, future: Future[_]): Boolean = {
ensureTypedActorEnabled
ensureEnabled
if (typedActorObjectInstance.get.isJoinPointAndOneWay(message)) {
future.asInstanceOf[CompletableFuture[Option[_]]].completeWithResult(None)
}
@ -173,7 +173,7 @@ object ReflectiveAccess extends Logging {
}
}
object EnterpriseModule {
object AkkaCloudModule {
type Mailbox = {
def enqueue(message: MessageInvocation)
@ -185,27 +185,27 @@ object ReflectiveAccess extends Logging {
def fromBinary(bytes: Array[Byte], clazz: Option[Class[_]]): AnyRef
}
lazy val isEnterpriseEnabled = clusterObjectInstance.isDefined
lazy val isEnabled = clusterObjectInstance.isDefined
val clusterObjectInstance: Option[AnyRef] =
getObjectFor("akka.cluster.Cluster$")
getObjectFor("akka.cloud.cluster.Cluster$")
val serializerClass: Option[Class[_]] =
getClassFor("akka.serialization.Serializer")
def ensureEnterpriseEnabled = if (!isEnterpriseEnabled) throw new ModuleNotAvailableException(
"Feature is only available in Akka Enterprise edition")
def ensureEnabled = if (!isEnabled) throw new ModuleNotAvailableException(
"Feature is only available in Akka Cloud")
def createFileBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.actor.mailbox.FileBasedMailbox", actorRef)
def createFileBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.FileBasedMailbox", actorRef)
def createZooKeeperBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.actor.mailbox.ZooKeeperBasedMailbox", actorRef)
def createZooKeeperBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.ZooKeeperBasedMailbox", actorRef)
def createBeanstalkBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.actor.mailbox.BeanstalkBasedMailbox", actorRef)
def createBeanstalkBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.BeanstalkBasedMailbox", actorRef)
def createRedisBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.actor.mailbox.RedisBasedMailbox", actorRef)
def createRedisBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.RedisBasedMailbox", actorRef)
private def createMailbox(mailboxClassname: String, actorRef: ActorRef): Mailbox = {
ensureEnterpriseEnabled
ensureEnabled
createInstance(
mailboxClassname,
Array(classOf[ActorRef]),

1
akka-actor/src/test/scala/akka/misc/ActorRegistrySpec.scala
View file

@ -27,7 +27,6 @@ object ActorRegistrySpec {
self.reply("got ping")
}
}
}
class ActorRegistrySpec extends JUnitSuite {

26
akka-typed-actor/src/test/scala/actor/typed-actor/TypedActorRegistrySpec.scala Normal file
View file

@ -0,0 +1,26 @@
package akka.actor
import org.scalatest.junit.JUnitSuite
import org.junit.Test
object TypedActorRegistrySpec {
trait My
class MyImpl extends TypedActor with My
}
class TypedActorRegistrySpec extends JUnitSuite {
import TypedActorRegistrySpec._
@Test def shouldGetTypedActorByClassFromActorRegistry {
ActorRegistry.shutdownAll
val my = TypedActor.newInstance[My](classOf[My], classOf[MyImpl], 3000)
val actors = ActorRegistry.typedActorsFor(classOf[My])
assert(actors.length === 1)
val option = ActorRegistry.typedActorFor[My]
assert(option != null)
assert(option.isDefined)
ActorRegistry.shutdownAll
}
}