Added methods for Cluster.remove and Cluster.release that accepts ActorRef
This commit is contained in:
Jonas Bonér
parent
5d4f8b4bcb
commit
df8c4dac89
3 changed files with 27 additions and 114 deletions
|
|
@ -299,6 +299,11 @@ trait ClusterNode {
|
|||
*/
|
||||
def store(actorRef: ActorRef, replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
|
||||
|
||||
/**
|
||||
* Removes actor from the cluster.
|
||||
*/
|
||||
def remove(actorRef: ActorRef)
|
||||
|
||||
/**
|
||||
* Removes actor with uuid from the cluster.
|
||||
*/
|
||||
|
|
@ -346,6 +351,11 @@ trait ClusterNode {
|
|||
*/
|
||||
def useActorOnNode(node: String, uuid: UUID)
|
||||
|
||||
/**
|
||||
* Checks in an actor after done using it on this node.
|
||||
*/
|
||||
def release(actorRef: ActorRef)
|
||||
|
||||
/**
|
||||
* Checks in an actor after done using it on this node.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -691,6 +691,13 @@ class DefaultClusterNode private[akka] (
|
|||
this
|
||||
} else throw new ClusterException("Not connected to cluster")
|
||||
|
||||
/**
|
||||
* Removes actor from the cluster.
|
||||
*/
|
||||
def remove(actorRef: ActorRef) {
|
||||
remove(actorRef.uuid)
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes actor with uuid from the cluster.
|
||||
*/
|
||||
|
|
@ -825,10 +832,20 @@ class DefaultClusterNode private[akka] (
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks in an actor after done using it on this node.
|
||||
*/
|
||||
def release(actorRef: ActorRef) {
|
||||
release(actorRef.address)
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks in an actor after done using it on this node.
|
||||
*/
|
||||
def release(actorAddress: String) {
|
||||
|
||||
// FIXME 'Cluster.release' needs to notify all existing ClusterActorRef's that are using the instance that it is no longer available. Then what to do? Should we even remove this method?
|
||||
|
||||
isConnected ifOn {
|
||||
actorUuidsForActorAddress(actorAddress) foreach { uuid ⇒
|
||||
EventHandler.debug(this,
|
||||
|
|
|
|||
|
|
@ -42,120 +42,6 @@ class ClusterSpec extends WordSpec with MustMatchers with BeforeAndAfterAll with
|
|||
|
||||
"A ClusterNode" should {
|
||||
|
||||
"be able to use an actor by actor address" in {
|
||||
val node = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "use-actor-id", port = 9001))
|
||||
node.start
|
||||
|
||||
// create actor
|
||||
val actorRef1 = actorOf[MyJavaSerializableActor]("actor-address").start
|
||||
(actorRef1 !! "hello").getOrElse("_") must equal("world 1")
|
||||
(actorRef1 !! "hello").getOrElse("_") must equal("world 2")
|
||||
|
||||
// register actor
|
||||
var serializeMailbox = true
|
||||
import BinaryFormatMyJavaSerializableActor._
|
||||
node.store(actorRef1, serializeMailbox)
|
||||
node.isClustered(actorRef1.address) must be(true)
|
||||
node.addressesForClusteredActors.exists(_ == actorRef1.address) must be(true)
|
||||
|
||||
// check out actor
|
||||
val actorRef2 = node.use(actorRef1.address).head
|
||||
node.isInUseOnNode(actorRef1.address, node = NodeAddress("test-cluster", "use-actor-id")) must be(true)
|
||||
(actorRef2 !! "hello").getOrElse("_") must equal("world 3")
|
||||
|
||||
actorRef1.stop
|
||||
actorRef2.stop
|
||||
|
||||
node.stop
|
||||
}
|
||||
|
||||
"be able to release an actor by address" in {
|
||||
val node = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "release-actor-id", port = 9001))
|
||||
node.start
|
||||
|
||||
// create actor
|
||||
val actorRef1 = actorOf[MyJavaSerializableActor]("actor-address").start
|
||||
(actorRef1 !! "hello").getOrElse("_") must equal("world 1")
|
||||
(actorRef1 !! "hello").getOrElse("_") must equal("world 2")
|
||||
|
||||
// register actor
|
||||
var serializeMailbox = true
|
||||
import BinaryFormatMyJavaSerializableActor._
|
||||
node.store(actorRef1, serializeMailbox)
|
||||
node.isClustered(actorRef1.address) must be(true)
|
||||
node.addressesForClusteredActors.exists(_ == actorRef1.address) must be(true)
|
||||
|
||||
// check out actor
|
||||
val actorRef2 = node.use(actorRef1.address).head
|
||||
node.isInUseOnNode(actorRef1.address, node = NodeAddress("test-cluster", "release-actor-id")) must be(true)
|
||||
(actorRef2 !! "hello").getOrElse("_") must equal("world 3")
|
||||
|
||||
// check in actor
|
||||
node.release(actorRef2.address)
|
||||
node.isInUseOnNode(actorRef1.address, node = NodeAddress("test-cluster", "release-actor-id")) must be(false)
|
||||
|
||||
actorRef1.stop
|
||||
actorRef2.stop
|
||||
|
||||
node.stop
|
||||
}
|
||||
|
||||
"be able to release used actor on remove an actor by actor address" in {
|
||||
// create actor
|
||||
val actorRef = actorOf[MyJavaSerializableActor]("actor-address").start
|
||||
|
||||
val node = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "remove-actor-id", port = 9001)).start
|
||||
val node2 = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "remove-actor-uuid-2", port = 9002)).start
|
||||
|
||||
// register actor
|
||||
import BinaryFormatMyJavaSerializableActor._
|
||||
var serializeMailbox = true
|
||||
node.store(actorRef, serializeMailbox)
|
||||
val actorRef2 = node2.use(actorRef.address).head
|
||||
|
||||
node2.isClustered(actorRef.address) must be(true)
|
||||
node.addressesForClusteredActors.exists(_ == actorRef.address) must be(true)
|
||||
node.nodesForActorsInUseWithAddress(actorRef.address) must have length (1)
|
||||
|
||||
// deregister actor
|
||||
node.remove(actorRef.address)
|
||||
node.addressesForClusteredActors.exists(_ == actorRef.address) must be(false)
|
||||
|
||||
node.nodesForActorsInUseWithAddress(actorRef.address) must have length (0)
|
||||
node.stop
|
||||
}
|
||||
|
||||
"be able to get home address for a clustered actor" in {
|
||||
val node = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "get-home-address", port = 9001))
|
||||
node.start
|
||||
|
||||
// create actor
|
||||
val actorRef1 = actorOf[MyJavaSerializableActor]("actor-address").start
|
||||
(actorRef1 !! "hello").getOrElse("_") must equal("world 1")
|
||||
(actorRef1 !! "hello").getOrElse("_") must equal("world 2")
|
||||
|
||||
// register actor
|
||||
var serializeMailbox = true
|
||||
import BinaryFormatMyJavaSerializableActor._
|
||||
node.store(actorRef1, serializeMailbox)
|
||||
node.isClustered(actorRef1.address) must be(true)
|
||||
node.uuidsForClusteredActors.exists(_ == actorRef1.uuid) must be(true)
|
||||
|
||||
// check out actor
|
||||
val actorRef2 = node.use(actorRef1.address).head
|
||||
node.isInUseOnNode(actorRef1.address, node = NodeAddress("test-cluster", "get-home-address")) must be(true)
|
||||
(actorRef2 !! "hello").getOrElse("_") must equal("world 3")
|
||||
|
||||
val addresses = node.addressesForActor(actorRef1.address)
|
||||
addresses.length must be > (0)
|
||||
addresses(0)._2.getPort must equal(9001)
|
||||
|
||||
actorRef1.stop
|
||||
actorRef2.stop
|
||||
|
||||
node.stop
|
||||
}
|
||||
|
||||
"be able to migrate an actor between two nodes using address" in {
|
||||
val node1 = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "migrate-id-1", port = 9001))
|
||||
val node2 = Cluster.newNode(nodeAddress = NodeAddress("test-cluster", "migrate-id-2", port = 9002))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue