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

Now requiring Cluster to be started and shut down manually when used outside of the Kernel

Browse source
This commit is contained in:
Viktor Klang 2010-02-03 20:55:42 +01:00
parent ba2b66bf55
commit fa009e4353
4 changed files with 61 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

66
akka-core/src/main/scala/remote/Cluster.scala
View file

@ -199,34 +199,40 @@ abstract class BasicClusterActor extends ClusterActor {
* Loads a specified ClusterActor and delegates to that instance.
*/
object Cluster extends Cluster with Logging {
private[remote] val clusterActor: Option[ClusterActor] = {
val name = config.getString("akka.remote.cluster.actor","not defined")
try {
val a = Class.forName(name).newInstance.asInstanceOf[ClusterActor]
a.start
Some(a)
}
catch {
case e => log.error(e,"Couldn't load Cluster provider: [%s]",name)
None
}
}
private[remote] val supervisor: Option[Supervisor] = if (clusterActor.isDefined) {
val sup = SupervisorFactory(
SupervisorConfig(
RestartStrategy(OneForOne, 5, 1000, List(classOf[Exception])),
Supervise(clusterActor.get, LifeCycle(Permanent)) :: Nil)
).newInstance
sup.start
Some(sup)
} else None
@volatile private[remote] var clusterActor: Option[ClusterActor] = None
@volatile private[remote] var supervisor: Option[Supervisor] = None
private[remote] lazy val serializer: Serializer = {
val className = config.getString("akka.remote.cluster.serializer", Serializer.Java.getClass.getName)
Class.forName(className).newInstance.asInstanceOf[Serializer]
}
private[remote] def createClusterActor : Option[ClusterActor] = {
val name = config.getString("akka.remote.cluster.actor")
try {
name map { fqn =>
val a = Class.forName(fqn).newInstance.asInstanceOf[ClusterActor]
a.start
a
}
}
catch {
case e => log.error(e,"Couldn't load Cluster provider: [%s]",name.getOrElse("Not specified")); None
}
}
private[remote] def createSupervisor(actor : ClusterActor) : Option[Supervisor] = {
val sup = SupervisorFactory(
SupervisorConfig(
RestartStrategy(OneForOne, 5, 1000, List(classOf[Exception])),
Supervise(actor, LifeCycle(Permanent)) :: Nil)
).newInstance
sup.start
Some(sup)
}
def name = clusterActor.map(_.name).getOrElse("No cluster")
def lookup[T](pf: PartialFunction[RemoteAddress, T]): Option[T] = clusterActor.flatMap(_.lookup(pf))
@ -237,5 +243,19 @@ object Cluster extends Cluster with Logging {
def relayMessage(to: Class[_ <: Actor], msg: AnyRef): Unit = clusterActor.foreach(_.relayMessage(to, msg))
def shutdown = supervisor.foreach(_.stop)
def start : Unit = synchronized {
if(supervisor.isEmpty) {
for(actor <- createClusterActor;
sup <- createSupervisor(actor)) {
clusterActor = Some(actor)
supervisor = Some(sup)
}
}
}
def shutdown : Unit = synchronized {
supervisor.foreach(_.stop)
supervisor = None
clusterActor = None
}
}