Moved the cluster impls to their own modules
This commit is contained in:
Viktor Klang
parent
9f12d05def
commit
46441e0010
8 changed files with 226 additions and 154 deletions
24
akka-cluster/akka-cluster-jgroups/pom.xml
Normal file
24
akka-cluster/akka-cluster-jgroups/pom.xml
Normal file
|
|
@ -0,0 +1,24 @@
|
||||||
|
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||||
|
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
|
||||||
|
<modelVersion>4.0.0</modelVersion>
|
||||||
|
|
||||||
|
<artifactId>akka-cluster-jgroups</artifactId>
|
||||||
|
<name>Akka Cluster JGroups Module</name>
|
||||||
|
|
||||||
|
<packaging>jar</packaging>
|
||||||
|
|
||||||
|
<parent>
|
||||||
|
<artifactId>akka-cluster-parent</artifactId>
|
||||||
|
<groupId>se.scalablesolutions.akka</groupId>
|
||||||
|
<version>0.6</version>
|
||||||
|
</parent>
|
||||||
|
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>jgroups</groupId>
|
||||||
|
<artifactId>jgroups</artifactId>
|
||||||
|
<version>2.8.0.CR7</version>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
|
||||||
|
</project>
|
||||||
|
|
@ -0,0 +1,155 @@
|
||||||
|
package se.scalablesolutions.akka.remote
|
||||||
|
|
||||||
|
import org.jgroups.{JChannel, View, Address, Message, ExtendedMembershipListener, Receiver}
|
||||||
|
|
||||||
|
import se.scalablesolutions.akka.config.ScalaConfig._
|
||||||
|
import se.scalablesolutions.akka.remote.Cluster.{Node, RelayedMessage}
|
||||||
|
import se.scalablesolutions.akka.actor.{ Actor, ActorRegistry}
|
||||||
|
|
||||||
|
import scala.collection.immutable.{Map, HashMap}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* JGroups Internal Cluster messages.
|
||||||
|
*/
|
||||||
|
private[remote] object JGroupsClusterActor {
|
||||||
|
sealed trait JGroupsClusterMessage
|
||||||
|
case object PapersPlease extends JGroupsClusterMessage
|
||||||
|
case class Papers(addresses: List[RemoteAddress]) extends JGroupsClusterMessage
|
||||||
|
case object Block extends JGroupsClusterMessage
|
||||||
|
case object Unblock extends JGroupsClusterMessage
|
||||||
|
case class Zombie(address: Address) extends JGroupsClusterMessage
|
||||||
|
case class RegisterLocalNode(server: RemoteAddress) extends JGroupsClusterMessage
|
||||||
|
case class DeregisterLocalNode(server: RemoteAddress) extends JGroupsClusterMessage
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Clustering support via JGroups.
|
||||||
|
*/
|
||||||
|
class JGroupsClusterActor extends ClusterActor {
|
||||||
|
import JGroupsClusterActor._
|
||||||
|
import org.scala_tools.javautils.Implicits._
|
||||||
|
|
||||||
|
|
||||||
|
@volatile private var isActive = false
|
||||||
|
@volatile private var local: Node = Node(Nil)
|
||||||
|
@volatile private var channel: Option[JChannel] = None
|
||||||
|
@volatile private var remotes: Map[Address, Node] = Map()
|
||||||
|
|
||||||
|
override def init = {
|
||||||
|
log debug "Initiating JGroups-based cluster actor"
|
||||||
|
remotes = new HashMap[Address, Node]
|
||||||
|
val me = this
|
||||||
|
isActive = true
|
||||||
|
|
||||||
|
// Set up the JGroups local endpoint
|
||||||
|
channel = Some(new JChannel {
|
||||||
|
setReceiver(new Receiver with ExtendedMembershipListener {
|
||||||
|
def getState: Array[Byte] = null
|
||||||
|
|
||||||
|
def setState(state: Array[Byte]): Unit = ()
|
||||||
|
|
||||||
|
def receive(msg: Message): Unit = if (isActive) me send msg
|
||||||
|
|
||||||
|
def viewAccepted(view: View): Unit = if (isActive) me send view
|
||||||
|
|
||||||
|
def suspect(a: Address): Unit = if (isActive) me send Zombie(a)
|
||||||
|
|
||||||
|
def block: Unit = if (isActive) me send Block
|
||||||
|
|
||||||
|
def unblock: Unit = if (isActive) me send Unblock
|
||||||
|
})
|
||||||
|
})
|
||||||
|
channel.map(_.connect(name))
|
||||||
|
}
|
||||||
|
|
||||||
|
def lookup[T](handleRemoteAddress: PartialFunction[RemoteAddress, T]): Option[T] =
|
||||||
|
remotes.values.toList.flatMap(_.endpoints).find(handleRemoteAddress isDefinedAt _).map(handleRemoteAddress)
|
||||||
|
|
||||||
|
def registerLocalNode(hostname: String, port: Int): Unit =
|
||||||
|
send(RegisterLocalNode(RemoteAddress(hostname, port)))
|
||||||
|
|
||||||
|
def deregisterLocalNode(hostname: String, port: Int): Unit =
|
||||||
|
send(DeregisterLocalNode(RemoteAddress(hostname, port)))
|
||||||
|
|
||||||
|
def relayMessage(to: Class[_ <: Actor], msg: AnyRef): Unit =
|
||||||
|
send(RelayedMessage(to.getName, msg))
|
||||||
|
|
||||||
|
private def broadcast[T <: AnyRef](recipients: Iterable[Address], msg: T): Unit = {
|
||||||
|
lazy val m = Cluster.serializer out msg
|
||||||
|
for (c <- channel; r <- recipients) c.send(new Message(r, null, m))
|
||||||
|
}
|
||||||
|
|
||||||
|
private def broadcast[T <: AnyRef](msg: T): Unit =
|
||||||
|
//Don't broadcast if we are not connected anywhere...
|
||||||
|
if (!remotes.isEmpty) channel.map(_.send(new Message(null, null, Cluster.serializer out msg)))
|
||||||
|
|
||||||
|
def receive = {
|
||||||
|
case Zombie(x) => { //Ask the presumed zombie for papers and prematurely treat it as dead
|
||||||
|
log debug ("Killing Zombie Node: %s", x)
|
||||||
|
broadcast(x :: Nil, PapersPlease)
|
||||||
|
remotes = remotes - x
|
||||||
|
}
|
||||||
|
|
||||||
|
case v: View => {
|
||||||
|
// Not present in the cluster anymore = presumably zombies
|
||||||
|
// Nodes we have no prior knowledge existed = unknowns
|
||||||
|
val members = Set[Address]() ++ v.getMembers.asScala - channel.get.getAddress // Exclude ourselves
|
||||||
|
val zombies = Set[Address]() ++ remotes.keySet -- members
|
||||||
|
val unknown = members -- remotes.keySet
|
||||||
|
|
||||||
|
log debug v.printDetails
|
||||||
|
|
||||||
|
// Tell the zombies and unknowns to provide papers and prematurely treat the zombies as dead
|
||||||
|
broadcast(zombies ++ unknown, PapersPlease)
|
||||||
|
remotes = remotes -- zombies
|
||||||
|
}
|
||||||
|
|
||||||
|
case m: Message => {
|
||||||
|
if (m.getSrc != channel.map(_.getAddress).getOrElse(m.getSrc)) // Handle non-own messages only, and only if we're connected
|
||||||
|
(Cluster.serializer in (m.getRawBuffer, None)) match {
|
||||||
|
|
||||||
|
case PapersPlease => {
|
||||||
|
log debug ("Asked for papers by %s", m.getSrc)
|
||||||
|
broadcast(m.getSrc :: Nil, Papers(local.endpoints))
|
||||||
|
|
||||||
|
if (remotes.get(m.getSrc).isEmpty) // If we were asked for papers from someone we don't know, ask them!
|
||||||
|
broadcast(m.getSrc :: Nil, PapersPlease)
|
||||||
|
}
|
||||||
|
|
||||||
|
case Papers(x) => remotes = remotes + (m.getSrc -> Node(x))
|
||||||
|
|
||||||
|
case RelayedMessage(c, m) => ActorRegistry.actorsFor(c).map(_ send m)
|
||||||
|
|
||||||
|
case unknown => log debug ("Unknown message: %s", unknown.toString)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
case rm @ RelayedMessage(_, _) => {
|
||||||
|
log debug ("Relaying message: %s", rm)
|
||||||
|
broadcast(rm)
|
||||||
|
}
|
||||||
|
|
||||||
|
case RegisterLocalNode(s) => {
|
||||||
|
log debug ("RegisterLocalNode: %s", s)
|
||||||
|
local = Node(local.endpoints + s)
|
||||||
|
broadcast(Papers(local.endpoints))
|
||||||
|
}
|
||||||
|
|
||||||
|
case DeregisterLocalNode(s) => {
|
||||||
|
log debug ("DeregisterLocalNode: %s", s)
|
||||||
|
local = Node(local.endpoints - s)
|
||||||
|
broadcast(Papers(local.endpoints))
|
||||||
|
}
|
||||||
|
|
||||||
|
case Block => log debug "UNSUPPORTED: JGroupsClusterActor::block" //TODO HotSwap to a buffering body
|
||||||
|
case Unblock => log debug "UNSUPPORTED: JGroupsClusterActor::unblock" //TODO HotSwap back and flush the buffer
|
||||||
|
}
|
||||||
|
|
||||||
|
override def shutdown = {
|
||||||
|
log debug ("Shutting down %s", toString)
|
||||||
|
isActive = false
|
||||||
|
channel.foreach(_.shutdown)
|
||||||
|
remotes = Map()
|
||||||
|
channel = None
|
||||||
|
}
|
||||||
|
}
|
||||||
40
akka-cluster/pom.xml
Normal file
40
akka-cluster/pom.xml
Normal file
|
|
@ -0,0 +1,40 @@
|
||||||
|
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||||
|
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
|
||||||
|
<modelVersion>4.0.0</modelVersion>
|
||||||
|
|
||||||
|
<artifactId>akka-cluster-parent</artifactId>
|
||||||
|
<name>Akka Cluster Modules</name>
|
||||||
|
|
||||||
|
<packaging>pom</packaging>
|
||||||
|
|
||||||
|
<parent>
|
||||||
|
<artifactId>akka</artifactId>
|
||||||
|
<groupId>se.scalablesolutions.akka</groupId>
|
||||||
|
<version>0.6</version>
|
||||||
|
</parent>
|
||||||
|
|
||||||
|
<modules>
|
||||||
|
<module>akka-cluster-jgroups</module>
|
||||||
|
</modules>
|
||||||
|
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<artifactId>akka-core</artifactId>
|
||||||
|
<groupId>${project.groupId}</groupId>
|
||||||
|
<version>${project.version}</version>
|
||||||
|
</dependency>
|
||||||
|
<!-- For Testing -->
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.scalatest</groupId>
|
||||||
|
<artifactId>scalatest</artifactId>
|
||||||
|
<version>1.0</version>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>junit</groupId>
|
||||||
|
<artifactId>junit</artifactId>
|
||||||
|
<version>4.5</version>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
</project>
|
||||||
|
|
@ -51,11 +51,6 @@
|
||||||
<artifactId>javautils</artifactId>
|
<artifactId>javautils</artifactId>
|
||||||
<version>2.7.4-0.1</version>
|
<version>2.7.4-0.1</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
<dependency>
|
|
||||||
<groupId>jgroups</groupId>
|
|
||||||
<artifactId>jgroups</artifactId>
|
|
||||||
<version>2.8.0.CR7</version>
|
|
||||||
</dependency>
|
|
||||||
|
|
||||||
<!-- For Protocol/Serialization -->
|
<!-- For Protocol/Serialization -->
|
||||||
<dependency>
|
<dependency>
|
||||||
|
|
|
||||||
|
|
@ -4,8 +4,6 @@
|
||||||
|
|
||||||
package se.scalablesolutions.akka.remote
|
package se.scalablesolutions.akka.remote
|
||||||
|
|
||||||
import org.jgroups.{JChannel, View, Address, Message, ExtendedMembershipListener, Receiver}
|
|
||||||
|
|
||||||
import se.scalablesolutions.akka.Config.config
|
import se.scalablesolutions.akka.Config.config
|
||||||
import se.scalablesolutions.akka.config.ScalaConfig._
|
import se.scalablesolutions.akka.config.ScalaConfig._
|
||||||
import se.scalablesolutions.akka.remote.Cluster.{Node, RelayedMessage}
|
import se.scalablesolutions.akka.remote.Cluster.{Node, RelayedMessage}
|
||||||
|
|
@ -83,149 +81,3 @@ object Cluster extends Cluster {
|
||||||
|
|
||||||
def shutdown = supervisor.foreach(_.stop)
|
def shutdown = supervisor.foreach(_.stop)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* JGroups Internal Cluster messages.
|
|
||||||
*/
|
|
||||||
private[remote] object JGroupsClusterActor {
|
|
||||||
sealed trait JGroupsClusterMessage
|
|
||||||
case object PapersPlease extends JGroupsClusterMessage
|
|
||||||
case class Papers(addresses: List[RemoteAddress]) extends JGroupsClusterMessage
|
|
||||||
case object Block extends JGroupsClusterMessage
|
|
||||||
case object Unblock extends JGroupsClusterMessage
|
|
||||||
case class Zombie(address: Address) extends JGroupsClusterMessage
|
|
||||||
case class RegisterLocalNode(server: RemoteAddress) extends JGroupsClusterMessage
|
|
||||||
case class DeregisterLocalNode(server: RemoteAddress) extends JGroupsClusterMessage
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Clustering support via JGroups.
|
|
||||||
*/
|
|
||||||
class JGroupsClusterActor extends ClusterActor {
|
|
||||||
import JGroupsClusterActor._
|
|
||||||
import org.scala_tools.javautils.Implicits._
|
|
||||||
|
|
||||||
|
|
||||||
@volatile private var isActive = false
|
|
||||||
@volatile private var local: Node = Node(Nil)
|
|
||||||
@volatile private var channel: Option[JChannel] = None
|
|
||||||
@volatile private var remotes: Map[Address, Node] = Map()
|
|
||||||
|
|
||||||
override def init = {
|
|
||||||
log debug "Initiating JGroups-based cluster actor"
|
|
||||||
remotes = new HashMap[Address, Node]
|
|
||||||
val me = this
|
|
||||||
isActive = true
|
|
||||||
|
|
||||||
// Set up the JGroups local endpoint
|
|
||||||
channel = Some(new JChannel {
|
|
||||||
setReceiver(new Receiver with ExtendedMembershipListener {
|
|
||||||
def getState: Array[Byte] = null
|
|
||||||
|
|
||||||
def setState(state: Array[Byte]): Unit = ()
|
|
||||||
|
|
||||||
def receive(msg: Message): Unit = if (isActive) me send msg
|
|
||||||
|
|
||||||
def viewAccepted(view: View): Unit = if (isActive) me send view
|
|
||||||
|
|
||||||
def suspect(a: Address): Unit = if (isActive) me send Zombie(a)
|
|
||||||
|
|
||||||
def block: Unit = if (isActive) me send Block
|
|
||||||
|
|
||||||
def unblock: Unit = if (isActive) me send Unblock
|
|
||||||
})
|
|
||||||
})
|
|
||||||
channel.map(_.connect(name))
|
|
||||||
}
|
|
||||||
|
|
||||||
def lookup[T](handleRemoteAddress: PartialFunction[RemoteAddress, T]): Option[T] =
|
|
||||||
remotes.values.toList.flatMap(_.endpoints).find(handleRemoteAddress isDefinedAt _).map(handleRemoteAddress)
|
|
||||||
|
|
||||||
def registerLocalNode(hostname: String, port: Int): Unit =
|
|
||||||
send(RegisterLocalNode(RemoteAddress(hostname, port)))
|
|
||||||
|
|
||||||
def deregisterLocalNode(hostname: String, port: Int): Unit =
|
|
||||||
send(DeregisterLocalNode(RemoteAddress(hostname, port)))
|
|
||||||
|
|
||||||
def relayMessage(to: Class[_ <: Actor], msg: AnyRef): Unit =
|
|
||||||
send(RelayedMessage(to.getName, msg))
|
|
||||||
|
|
||||||
private def broadcast[T <: AnyRef](recipients: Iterable[Address], msg: T): Unit = {
|
|
||||||
lazy val m = Cluster.serializer out msg
|
|
||||||
for (c <- channel; r <- recipients) c.send(new Message(r, null, m))
|
|
||||||
}
|
|
||||||
|
|
||||||
private def broadcast[T <: AnyRef](msg: T): Unit =
|
|
||||||
//Don't broadcast if we are not connected anywhere...
|
|
||||||
if (!remotes.isEmpty) channel.map(_.send(new Message(null, null, Cluster.serializer out msg)))
|
|
||||||
|
|
||||||
def receive = {
|
|
||||||
case Zombie(x) => { //Ask the presumed zombie for papers and prematurely treat it as dead
|
|
||||||
log debug ("Killing Zombie Node: %s", x)
|
|
||||||
broadcast(x :: Nil, PapersPlease)
|
|
||||||
remotes = remotes - x
|
|
||||||
}
|
|
||||||
|
|
||||||
case v: View => {
|
|
||||||
// Not present in the cluster anymore = presumably zombies
|
|
||||||
// Nodes we have no prior knowledge existed = unknowns
|
|
||||||
val members = Set[Address]() ++ v.getMembers.asScala - channel.get.getAddress // Exclude ourselves
|
|
||||||
val zombies = Set[Address]() ++ remotes.keySet -- members
|
|
||||||
val unknown = members -- remotes.keySet
|
|
||||||
|
|
||||||
log debug v.printDetails
|
|
||||||
|
|
||||||
// Tell the zombies and unknowns to provide papers and prematurely treat the zombies as dead
|
|
||||||
broadcast(zombies ++ unknown, PapersPlease)
|
|
||||||
remotes = remotes -- zombies
|
|
||||||
}
|
|
||||||
|
|
||||||
case m: Message => {
|
|
||||||
if (m.getSrc != channel.map(_.getAddress).getOrElse(m.getSrc)) // Handle non-own messages only, and only if we're connected
|
|
||||||
(Cluster.serializer in (m.getRawBuffer, None)) match {
|
|
||||||
|
|
||||||
case PapersPlease => {
|
|
||||||
log debug ("Asked for papers by %s", m.getSrc)
|
|
||||||
broadcast(m.getSrc :: Nil, Papers(local.endpoints))
|
|
||||||
|
|
||||||
if (remotes.get(m.getSrc).isEmpty) // If we were asked for papers from someone we don't know, ask them!
|
|
||||||
broadcast(m.getSrc :: Nil, PapersPlease)
|
|
||||||
}
|
|
||||||
|
|
||||||
case Papers(x) => remotes = remotes + (m.getSrc -> Node(x))
|
|
||||||
|
|
||||||
case RelayedMessage(c, m) => ActorRegistry.actorsFor(c).map(_ send m)
|
|
||||||
|
|
||||||
case unknown => log debug ("Unknown message: %s", unknown.toString)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
case rm @ RelayedMessage(_, _) => {
|
|
||||||
log debug ("Relaying message: %s", rm)
|
|
||||||
broadcast(rm)
|
|
||||||
}
|
|
||||||
|
|
||||||
case RegisterLocalNode(s) => {
|
|
||||||
log debug ("RegisterLocalNode: %s", s)
|
|
||||||
local = Node(local.endpoints + s)
|
|
||||||
broadcast(Papers(local.endpoints))
|
|
||||||
}
|
|
||||||
|
|
||||||
case DeregisterLocalNode(s) => {
|
|
||||||
log debug ("DeregisterLocalNode: %s", s)
|
|
||||||
local = Node(local.endpoints - s)
|
|
||||||
broadcast(Papers(local.endpoints))
|
|
||||||
}
|
|
||||||
|
|
||||||
case Block => log debug "UNSUPPORTED: JGroupsClusterActor::block" //TODO HotSwap to a buffering body
|
|
||||||
case Unblock => log debug "UNSUPPORTED: JGroupsClusterActor::unblock" //TODO HotSwap back and flush the buffer
|
|
||||||
}
|
|
||||||
|
|
||||||
override def shutdown = {
|
|
||||||
log debug ("Shutting down %s", toString)
|
|
||||||
isActive = false
|
|
||||||
channel.foreach(_.shutdown)
|
|
||||||
remotes = Map()
|
|
||||||
channel = None
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -51,6 +51,11 @@
|
||||||
<groupId>${project.groupId}</groupId>
|
<groupId>${project.groupId}</groupId>
|
||||||
<version>${project.version}</version>
|
<version>${project.version}</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<artifactId>akka-cluster-jgroups</artifactId>
|
||||||
|
<groupId>${project.groupId}</groupId>
|
||||||
|
<version>${project.version}</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
<!-- For Atmosphere -->
|
<!-- For Atmosphere -->
|
||||||
<dependency>
|
<dependency>
|
||||||
|
|
|
||||||
|
|
@ -48,7 +48,7 @@
|
||||||
|
|
||||||
<cluster>
|
<cluster>
|
||||||
name = "default" # The name of the cluster
|
name = "default" # The name of the cluster
|
||||||
actor = "se.scalablesolutions.akka.remote.JGroupsClusterActor" # FQN of an implementation of ClusterActor
|
#actor = "se.scalablesolutions.akka.remote.JGroupsClusterActor" # FQN of an implementation of ClusterActor
|
||||||
serializer = "se.scalablesolutions.akka.serialization.Serializer$Java" # FQN of the serializer class
|
serializer = "se.scalablesolutions.akka.serialization.Serializer$Java" # FQN of the serializer class
|
||||||
</cluster>
|
</cluster>
|
||||||
|
|
||||||
|
|
|
||||||
1
pom.xml
1
pom.xml
|
|
@ -52,6 +52,7 @@
|
||||||
<module>akka-util</module>
|
<module>akka-util</module>
|
||||||
<module>akka-core</module>
|
<module>akka-core</module>
|
||||||
<module>akka-persistence</module>
|
<module>akka-persistence</module>
|
||||||
|
<module>akka-cluster</module>
|
||||||
<module>akka-rest</module>
|
<module>akka-rest</module>
|
||||||
<module>akka-comet</module>
|
<module>akka-comet</module>
|
||||||
<module>akka-amqp</module>
|
<module>akka-amqp</module>
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue