Renamed 'replication-factor' config element to 'nr-of-instances' and 'ReplicationFactor' case class to 'NrOfInstances'.

Signed-off-by: Jonas Bonér <jonas@jonasboner.com>
This commit is contained in:
Jonas Bonér 2011-09-28 19:42:12 +02:00
parent 16e4be6077
commit 0957e41d19
39 changed files with 140 additions and 140 deletions

View file

@ -20,12 +20,12 @@ class DeployerSpec extends WordSpec with MustMatchers {
"service-ping",
None,
LeastCPU,
ReplicationFactor(3),
NrOfInstances(3),
BannagePeriodFailureDetector(10),
RemoteScope("localhost", 2552))))
// ClusterScope(
// List(Node("node1")),
// new ReplicationFactor(3),
// new NrOfInstances(3),
// Replication(
// TransactionLog,
// WriteThrough)))))

View file

@ -24,7 +24,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
// address,
// None,
// Direct,
// ReplicationFactor(1),
// NrOfInstances(1),
// RemoveConnectionOnFirstFailureLocalFailureDetector,
// LocalScope))
@ -58,7 +58,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
// address,
// None,
// Direct,
// ReplicationFactor(1),
// NrOfInstances(1),
// RemoveConnectionOnFirstFailureLocalFailureDetector,
// LocalScope))
@ -88,7 +88,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
// address,
// None,
// Direct,
// ReplicationFactor(1),
// NrOfInstances(1),
// RemoveConnectionOnFirstFailureLocalFailureDetector,
// LocalScope))
@ -121,7 +121,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
address,
None,
RoundRobin,
ReplicationFactor(5),
NrOfInstances(5),
RemoveConnectionOnFirstFailureLocalFailureDetector,
LocalScope))
@ -157,7 +157,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
address,
None,
RoundRobin,
ReplicationFactor(10),
NrOfInstances(10),
RemoveConnectionOnFirstFailureLocalFailureDetector,
LocalScope))
@ -202,7 +202,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
address,
None,
RoundRobin,
ReplicationFactor(5),
NrOfInstances(5),
RemoveConnectionOnFirstFailureLocalFailureDetector,
LocalScope))
@ -237,7 +237,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
address,
None,
Random,
ReplicationFactor(7),
NrOfInstances(7),
RemoveConnectionOnFirstFailureLocalFailureDetector,
LocalScope))
@ -271,7 +271,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
address,
None,
Random,
ReplicationFactor(10),
NrOfInstances(10),
RemoveConnectionOnFirstFailureLocalFailureDetector,
LocalScope))
@ -316,7 +316,7 @@ class ConfiguredLocalRoutingSpec extends WordSpec with MustMatchers {
address,
None,
Random,
ReplicationFactor(6),
NrOfInstances(6),
RemoveConnectionOnFirstFailureLocalFailureDetector,
LocalScope))

View file

@ -122,7 +122,7 @@ object Deployer extends ActorDeployer {
val addressPath = "akka.actor.deployment." + address
configuration.getSection(addressPath) match {
case None
Some(Deploy(address, None, Direct, ReplicationFactor(1), RemoveConnectionOnFirstFailureLocalFailureDetector, LocalScope))
Some(Deploy(address, None, Direct, NrOfInstances(1), RemoveConnectionOnFirstFailureLocalFailureDetector, LocalScope))
case Some(addressConfig)
@ -145,22 +145,23 @@ object Deployer extends ActorDeployer {
}
// --------------------------------
// akka.actor.deployment.<address>.replication-factor
// akka.actor.deployment.<address>.nr-of-instances
// --------------------------------
val nrOfInstances = {
if (router == Direct) new ReplicationFactor(1)
if (router == Direct) NrOfInstances(1)
else {
addressConfig.getAny("replication-factor", "0") match {
case "auto" AutoReplicationFactor
case "0" ZeroReplicationFactor
addressConfig.getAny("nr-of-instances", "1") match {
case "auto" AutoNrOfInstances
case "1" NrOfInstances(1)
case "0" ZeroNrOfInstances
case nrOfReplicas: String
try {
new ReplicationFactor(nrOfReplicas.toInt)
new NrOfInstances(nrOfReplicas.toInt)
} catch {
case e: Exception
throw new ConfigurationException(
"Config option [" + addressPath +
".cluster.replication-factor] needs to be either [\"auto\"] or [0-N] - was [" +
".nr-of-instances] needs to be either [\"auto\"] or [1-N] - was [" +
nrOfReplicas + "]")
}
}

View file

@ -23,7 +23,7 @@ object DeploymentConfig {
address: String,
recipe: Option[ActorRecipe],
routing: Routing = Direct,
nrOfInstances: ReplicationFactor = ZeroReplicationFactor,
nrOfInstances: NrOfInstances = ZeroNrOfInstances,
failureDetector: FailureDetector = RemoveConnectionOnFirstFailureLocalFailureDetector,
scope: Scope = LocalScope) {
Address.validate(address)
@ -101,28 +101,28 @@ object DeploymentConfig {
// --- Replicas
// --------------------------------
class ReplicationFactor(val factor: Int) extends Serializable {
if (factor < 0) throw new IllegalArgumentException("replication-factor can not be negative")
class NrOfInstances(val factor: Int) extends Serializable {
if (factor < 0) throw new IllegalArgumentException("nr-of-instances can not be negative")
override def hashCode = 0 + factor.##
override def equals(other: Any) = ReplicationFactor.unapply(this) == ReplicationFactor.unapply(other)
override def toString = "ReplicationFactor(" + factor + ")"
override def equals(other: Any) = NrOfInstances.unapply(this) == NrOfInstances.unapply(other)
override def toString = "NrOfInstances(" + factor + ")"
}
object ReplicationFactor {
def apply(factor: Int): ReplicationFactor = new ReplicationFactor(factor)
object NrOfInstances {
def apply(factor: Int): NrOfInstances = new NrOfInstances(factor)
def unapply(other: Any) = other match {
case x: ReplicationFactor import x._; Some(factor)
case _ None
case x: NrOfInstances import x._; Some(factor)
case _ None
}
}
// For Java API
class AutoReplicationFactor extends ReplicationFactor(-1)
class ZeroReplicationFactor extends ReplicationFactor(0)
class AutoNrOfInstances extends NrOfInstances(-1)
class ZeroNrOfInstances extends NrOfInstances(0)
// For Scala API
case object AutoReplicationFactor extends AutoReplicationFactor
case object ZeroReplicationFactor extends ZeroReplicationFactor
case object AutoNrOfInstances extends AutoNrOfInstances
case object ZeroNrOfInstances extends ZeroNrOfInstances
// --------------------------------
// --- Replication

View file

@ -248,14 +248,14 @@ trait ClusterNode {
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, serializer: Serializer): ClusterNode
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, serializer: Serializer): ClusterNode
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@ -276,14 +276,14 @@ trait ClusterNode {
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](address: String, actorClass: Class[T], replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
def store[T <: Actor](address: String, actorClass: Class[T], nrOfInstances: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
@ -311,14 +311,14 @@ trait ClusterNode {
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, serializer: Serializer): ClusterNode
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, serializer: Serializer): ClusterNode
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
@ -329,26 +329,26 @@ trait ClusterNode {
/**
* Needed to have reflection through structural typing work.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode
/**
* Needed to have reflection through structural typing work.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode
/**
* Removes actor from the cluster.
@ -442,28 +442,28 @@ trait ClusterNode {
def inetSocketAddressesForActor(actorAddress: String): Array[(UUID, InetSocketAddress)]
/**
* Send a function 'Function0[Unit]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument).
* Send a function 'Function0[Unit]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument).
*/
def send(f: Function0[Unit], replicationFactor: Int)
def send(f: Function0[Unit], nrOfInstances: Int)
/**
* Send a function 'Function0[Any]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument).
* Send a function 'Function0[Any]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument).
* Returns an 'Array' with all the 'Future's from the computation.
*/
def send(f: Function0[Any], replicationFactor: Int): List[Future[Any]]
def send(f: Function0[Any], nrOfInstances: Int): List[Future[Any]]
/**
* Send a function 'Function1[Any, Unit]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument)
* Send a function 'Function1[Any, Unit]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument)
* with the argument speficied.
*/
def send(f: Function1[Any, Unit], arg: Any, replicationFactor: Int)
def send(f: Function1[Any, Unit], arg: Any, nrOfInstances: Int)
/**
* Send a function 'Function1[Any, Any]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument)
* Send a function 'Function1[Any, Any]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument)
* with the argument speficied.
* Returns an 'Array' with all the 'Future's from the computation.
*/
def send(f: Function1[Any, Any], arg: Any, replicationFactor: Int): List[Future[Any]]
def send(f: Function1[Any, Any], arg: Any, nrOfInstances: Int): List[Future[Any]]
/**
* Stores a configuration element under a specific key.

View file

@ -517,16 +517,16 @@ class DefaultClusterNode private[akka] (
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), replicationFactor, Transient, false, serializer)
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), nrOfInstances, Transient, false, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), replicationFactor, replicationScheme, false, serializer)
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), nrOfInstances, replicationScheme, false, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
@ -549,16 +549,16 @@ class DefaultClusterNode private[akka] (
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), replicationFactor, Transient, serializeMailbox, serializer)
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), nrOfInstances, Transient, serializeMailbox, serializer)
/**
* Clusters an actor of a specific type. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store[T <: Actor](actorAddress: String, actorClass: Class[T], replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), replicationFactor, replicationScheme, serializeMailbox, serializer)
def store[T <: Actor](actorAddress: String, actorClass: Class[T], nrOfInstances: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
store(actorAddress, () Actor.actorOf(actorClass, actorAddress), nrOfInstances, replicationScheme, serializeMailbox, serializer)
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
@ -589,24 +589,24 @@ class DefaultClusterNode private[akka] (
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, serializer: Serializer): ClusterNode =
store(actorAddress, actorFactory, replicationFactor, Transient, false, serializer)
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, serializer: Serializer): ClusterNode =
store(actorAddress, actorFactory, nrOfInstances, Transient, false, serializer)
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode =
store(actorAddress, actorFactory, replicationFactor, replicationScheme, false, serializer)
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, replicationScheme: ReplicationScheme, serializer: Serializer): ClusterNode =
store(actorAddress, actorFactory, nrOfInstances, replicationScheme, false, serializer)
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
* with the actor passed in as argument. You can use this to save off snapshots of the actor to a highly
* available durable store.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
store(actorAddress, actorFactory, replicationFactor, Transient, serializeMailbox, serializer)
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, serializeMailbox: Boolean, serializer: Serializer): ClusterNode =
store(actorAddress, actorFactory, nrOfInstances, Transient, serializeMailbox, serializer)
/**
* Clusters an actor with UUID. If the actor is already clustered then the clustered version will be updated
@ -619,14 +619,14 @@ class DefaultClusterNode private[akka] (
/**
* Needed to have reflection through structural typing work.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode =
store(actorAddress, actorFactory, replicationFactor, replicationScheme, serializeMailbox, serializer.asInstanceOf[Serializer])
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, replicationScheme: ReplicationScheme, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode =
store(actorAddress, actorFactory, nrOfInstances, replicationScheme, serializeMailbox, serializer.asInstanceOf[Serializer])
/**
* Needed to have reflection through structural typing work.
*/
def store(actorAddress: String, actorFactory: () ActorRef, replicationFactor: Int, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode =
store(actorAddress, actorFactory, replicationFactor, Transient, serializeMailbox, serializer)
def store(actorAddress: String, actorFactory: () ActorRef, nrOfInstances: Int, serializeMailbox: Boolean, serializer: AnyRef): ClusterNode =
store(actorAddress, actorFactory, nrOfInstances, Transient, serializeMailbox, serializer)
/**
* Clusters an actor. If the actor is already clustered then the clustered version will be updated
@ -636,7 +636,7 @@ class DefaultClusterNode private[akka] (
def store(
actorAddress: String,
actorFactory: () ActorRef,
replicationFactor: Int,
nrOfInstances: Int,
replicationScheme: ReplicationScheme,
serializeMailbox: Boolean,
serializer: Serializer): ClusterNode = {
@ -686,7 +686,7 @@ class DefaultClusterNode private[akka] (
// create ADDRESS -> UUIDs mapping
ignore[ZkNodeExistsException](zkClient.createPersistent(actorAddressToUuidsPathFor(actorAddress)))
useActorOnNodes(nodesForReplicationFactor(replicationFactor, Some(actorAddress)).toArray, actorAddress)
useActorOnNodes(nodesForNrOfInstances(nrOfInstances, Some(actorAddress)).toArray, actorAddress)
this
}
@ -1025,9 +1025,9 @@ class DefaultClusterNode private[akka] (
// =======================================
/**
* Send a function 'Function0[Unit]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument).
* Send a function 'Function0[Unit]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument).
*/
def send(f: Function0[Unit], replicationFactor: Int) {
def send(f: Function0[Unit], nrOfInstances: Int) {
Serialization.serialize(f) match {
case Left(error) throw error
case Right(bytes)
@ -1035,15 +1035,15 @@ class DefaultClusterNode private[akka] (
.setMessageType(FUNCTION_FUN0_UNIT)
.setPayload(ByteString.copyFrom(bytes))
.build
nodeConnectionsForReplicationFactor(replicationFactor) foreach (_ ! message)
nodeConnectionsForNrOfInstances(nrOfInstances) foreach (_ ! message)
}
}
/**
* Send a function 'Function0[Any]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument).
* Send a function 'Function0[Any]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument).
* Returns an 'Array' with all the 'Future's from the computation.
*/
def send(f: Function0[Any], replicationFactor: Int): List[Future[Any]] = {
def send(f: Function0[Any], nrOfInstances: Int): List[Future[Any]] = {
Serialization.serialize(f) match {
case Left(error) throw error
case Right(bytes)
@ -1051,16 +1051,16 @@ class DefaultClusterNode private[akka] (
.setMessageType(FUNCTION_FUN0_ANY)
.setPayload(ByteString.copyFrom(bytes))
.build
val results = nodeConnectionsForReplicationFactor(replicationFactor) map (_ ? message)
val results = nodeConnectionsForNrOfInstances(nrOfInstances) map (_ ? message)
results.toList.asInstanceOf[List[Future[Any]]]
}
}
/**
* Send a function 'Function1[Any, Unit]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument)
* Send a function 'Function1[Any, Unit]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument)
* with the argument speficied.
*/
def send(f: Function1[Any, Unit], arg: Any, replicationFactor: Int) {
def send(f: Function1[Any, Unit], arg: Any, nrOfInstances: Int) {
Serialization.serialize((f, arg)) match {
case Left(error) throw error
case Right(bytes)
@ -1068,16 +1068,16 @@ class DefaultClusterNode private[akka] (
.setMessageType(FUNCTION_FUN1_ARG_UNIT)
.setPayload(ByteString.copyFrom(bytes))
.build
nodeConnectionsForReplicationFactor(replicationFactor) foreach (_ ! message)
nodeConnectionsForNrOfInstances(nrOfInstances) foreach (_ ! message)
}
}
/**
* Send a function 'Function1[Any, Any]' to be invoked on a random number of nodes (defined by 'replicationFactor' argument)
* Send a function 'Function1[Any, Any]' to be invoked on a random number of nodes (defined by 'nrOfInstances' argument)
* with the argument speficied.
* Returns an 'Array' with all the 'Future's from the computation.
*/
def send(f: Function1[Any, Any], arg: Any, replicationFactor: Int): List[Future[Any]] = {
def send(f: Function1[Any, Any], arg: Any, nrOfInstances: Int): List[Future[Any]] = {
Serialization.serialize((f, arg)) match {
case Left(error) throw error
case Right(bytes)
@ -1085,7 +1085,7 @@ class DefaultClusterNode private[akka] (
.setMessageType(FUNCTION_FUN1_ARG_ANY)
.setPayload(ByteString.copyFrom(bytes))
.build
val results = nodeConnectionsForReplicationFactor(replicationFactor) map (_ ? message)
val results = nodeConnectionsForNrOfInstances(nrOfInstances) map (_ ? message)
results.toList.asInstanceOf[List[Future[Any]]]
}
}
@ -1211,16 +1211,16 @@ class DefaultClusterNode private[akka] (
private[cluster] def actorAddressToUuidsPathFor(actorAddress: String, uuid: UUID): String = "%s/%s".format(actorAddressToUuidsPathFor(actorAddress), uuid)
/**
* Returns a random set with node names of size 'replicationFactor'.
* Default replicationFactor is 0, which returns the empty Set.
* Returns a random set with node names of size 'nrOfInstances'.
* Default nrOfInstances is 0, which returns the empty Set.
*/
private def nodesForReplicationFactor(replicationFactor: Int = 0, actorAddress: Option[String] = None): Set[String] = {
private def nodesForNrOfInstances(nrOfInstances: Int = 0, actorAddress: Option[String] = None): Set[String] = {
var replicaNames = Set.empty[String]
val nrOfClusterNodes = nodeConnections.get.connections.size
if (replicationFactor < 1) return replicaNames
if (nrOfClusterNodes < replicationFactor) throw new IllegalArgumentException(
"Replication factor [" + replicationFactor +
if (nrOfInstances < 1) return replicaNames
if (nrOfClusterNodes < nrOfInstances) throw new IllegalArgumentException(
"Replication factor [" + nrOfInstances +
"] is greater than the number of available nodeNames [" + nrOfClusterNodes + "]")
val preferredNodes =
@ -1228,7 +1228,7 @@ class DefaultClusterNode private[akka] (
// use 'preferred-nodes' in deployment config for the actor
Deployer.deploymentFor(actorAddress.get) match {
case Deploy(_, _, _, _, Cluster(nodes, _, _))
nodes map (node DeploymentConfig.nodeNameFor(node)) take replicationFactor
nodes map (node DeploymentConfig.nodeNameFor(node)) take nrOfInstances
case _
throw new ClusterException("Actor [" + actorAddress.get + "] is not configured as clustered")
}
@ -1243,11 +1243,11 @@ class DefaultClusterNode private[akka] (
val nrOfCurrentReplicaNames = replicaNames.size
val replicaSet =
if (nrOfCurrentReplicaNames > replicationFactor) throw new IllegalStateException("Replica set is larger than replication factor")
else if (nrOfCurrentReplicaNames == replicationFactor) replicaNames
if (nrOfCurrentReplicaNames > nrOfInstances) throw new IllegalStateException("Replica set is larger than replication factor")
else if (nrOfCurrentReplicaNames == nrOfInstances) replicaNames
else {
val random = new java.util.Random(System.currentTimeMillis)
while (replicaNames.size < replicationFactor) {
while (replicaNames.size < nrOfInstances) {
replicaNames = replicaNames + membershipNodes(random.nextInt(nrOfClusterNodes))
}
replicaNames
@ -1260,12 +1260,12 @@ class DefaultClusterNode private[akka] (
}
/**
* Returns a random set with replica connections of size 'replicationFactor'.
* Default replicationFactor is 0, which returns the empty Set.
* Returns a random set with replica connections of size 'nrOfInstances'.
* Default nrOfInstances is 0, which returns the empty Set.
*/
private def nodeConnectionsForReplicationFactor(replicationFactor: Int = 0, actorAddress: Option[String] = None): Set[ActorRef] = {
private def nodeConnectionsForNrOfInstances(nrOfInstances: Int = 0, actorAddress: Option[String] = None): Set[ActorRef] = {
for {
node nodesForReplicationFactor(replicationFactor, actorAddress)
node nodesForNrOfInstances(nrOfInstances, actorAddress)
connectionOption nodeConnections.get.connections(node)
connection connectionOption
actorRef connection._2

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 1
akka.actor.deployment.service-hello.nr-of-instances = 1

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 1
akka.actor.deployment.service-hello.nr-of-instances = 1

View file

@ -3,5 +3,5 @@ akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-test.router = "round-robin"
akka.actor.deployment.service-test.cluster.preferred-nodes = ["node:node2","node:node3"]
akka.actor.deployment.service-test.cluster.replication-factor = 2
akka.actor.deployment.service-test.nr-of-instances = 2
akka.remote.client.buffering.retry-message-send-on-failure = false

View file

@ -2,5 +2,5 @@ akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-test.router = "round-robin"
akka.actor.deployment.service-test.cluster.preferred-nodes = ["node:node2","node:node3"]
akka.actor.deployment.service-test.cluster.replication-factor = 2
akka.actor.deployment.service-test.nr-of-instances = 2
akka.remote.client.buffering.retry-message-send-on-failure = false

View file

@ -2,5 +2,5 @@ akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-test.router = "round-robin"
akka.actor.deployment.service-test.cluster.preferred-nodes = ["node:node2","node:node3"]
akka.actor.deployment.service-test.cluster.replication-factor = 2
akka.actor.deployment.service-test.nr-of-instances = 2
akka.remote.client.buffering.retry-message-send-on-failure = false

View file

@ -1,7 +1,7 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.hello-world-write-behind-nosnapshot.router = "direct"
akka.actor.deployment.hello-world-write-behind-nosnapshot.cluster.replication-factor = 1
akka.actor.deployment.hello-world-write-behind-nosnapshot.nr-of-instances = 1
akka.actor.deployment.hello-world-write-behind-nosnapshot.cluster.replication.storage = "transaction-log"
akka.actor.deployment.hello-world-write-behind-nosnapshot.cluster.replication.strategy = "write-behind"
akka.cluster.replication.snapshot-frequency = 1000

View file

@ -1,7 +1,7 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.hello-world-write-behind-nosnapshot.router = "direct"
akka.actor.deployment.hello-world-write-behind-nosnapshot.cluster.replication-factor = 1
akka.actor.deployment.hello-world-write-behind-nosnapshot.nr-of-instances = 1
akka.actor.deployment.hello-world-write-behind-nosnapshot.cluster.replication.storage = "transaction-log"
akka.actor.deployment.hello-world-write-behind-nosnapshot.cluster.replication.strategy = "write-behind"
akka.cluster.replication.snapshot-frequency = 1000

View file

@ -1,7 +1,7 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.hello-world.router = "direct"
akka.actor.deployment.hello-world-write-behind-snapshot.cluster.replication-factor = 1
akka.actor.deployment.hello-world-write-behind-snapshot.nr-of-instances = 1
akka.actor.deployment.hello-world-write-behind-snapshot.cluster.replication.storage = "transaction-log"
akka.actor.deployment.hello-world-write-behind-snapshot.cluster.replication.strategy = "write-behind"
akka.cluster.replication.snapshot-frequency = 7

View file

@ -1,7 +1,7 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.hello-world.router = "direct"
akka.actor.deployment.hello-world-write-behind-snapshot.cluster.replication-factor = 1
akka.actor.deployment.hello-world-write-behind-snapshot.nr-of-instances = 1
akka.actor.deployment.hello-world-write-behind-snapshot.cluster.replication.storage = "transaction-log"
akka.actor.deployment.hello-world-write-behind-snapshot.cluster.replication.strategy = "write-behind"
akka.cluster.replication.snapshot-frequency = 7

View file

@ -1,7 +1,7 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.hello-world-write-through-snapshot.router = "direct"
akka.actor.deployment.hello-world-write-through-snapshot.cluster.replication-factor = 1
akka.actor.deployment.hello-world-write-through-snapshot.nr-of-instances = 1
akka.actor.deployment.hello-world-write-through-snapshot.cluster.replication.storage = "transaction-log"
akka.actor.deployment.hello-world-write-through-snapshot.cluster.replication.strategy = "write-through"
akka.cluster.replication.snapshot-frequency = 7

View file

@ -1,7 +1,7 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.hello-world-write-through-snapshot.router = "direct"
akka.actor.deployment.hello-world-write-through-snapshot.cluster.replication-factor = 1
akka.actor.deployment.hello-world-write-through-snapshot.nr-of-instances = 1
akka.actor.deployment.hello-world-write-through-snapshot.cluster.replication.storage = "transaction-log"
akka.actor.deployment.hello-world-write-through-snapshot.cluster.replication.strategy = "write-through"
akka.cluster.replication.snapshot-frequency = 7

View file

@ -3,6 +3,6 @@ akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "random"
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1", "node:node3"]
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.timeout = 30
akka.cluster.session-timeout = 10

View file

@ -3,6 +3,6 @@ akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "random"
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1", "node:node3"]
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.timeout = 30
akka.cluster.session-timeout = 10

View file

@ -3,6 +3,6 @@ akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "random"
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1", "node:node3"]
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.timeout = 30
akka.cluster.session-timeout = 10

View file

@ -2,7 +2,7 @@ akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-node1.router = "random"
akka.actor.deployment.service-node1.cluster.preferred-nodes = ["node:node1"]
akka.actor.deployment.service-node1.cluster.replication-factor = 1
akka.actor.deployment.service-node1.nr-of-instances = 1
akka.actor.deployment.service-node2.router = "random"
akka.actor.deployment.service-node2.cluster.preferred-nodes = ["node:node2"]
akka.actor.deployment.service-node2.cluster.replication-factor = 1
akka.actor.deployment.service-node2.nr-of-instances = 1

View file

@ -2,7 +2,7 @@ akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-node1.router = "random"
akka.actor.deployment.service-node1.cluster.preferred-nodes = ["node:node1"]
akka.actor.deployment.service-node1.cluster.replication-factor = 1
akka.actor.deployment.service-node1.nr-of-instances = 1
akka.actor.deployment.service-node2.router = "random"
akka.actor.deployment.service-node2.cluster.preferred-nodes = ["node:node2"]
akka.actor.deployment.service-node2.cluster.replication-factor = 1
akka.actor.deployment.service-node2.nr-of-instances = 1

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "random"
akka.actor.deployment.service-hello.cluster.replication-factor = 1
akka.actor.deployment.service-hello.nr-of-instances = 1

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "random"
akka.actor.deployment.service-hello.cluster.replication-factor = 3
akka.actor.deployment.service-hello.nr-of-instances = 3

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "random"
akka.actor.deployment.service-hello.cluster.replication-factor = 3
akka.actor.deployment.service-hello.nr-of-instances = 3

View file

@ -2,7 +2,7 @@ akka.enabled-modules = ["cluster"]
akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1","node:node3"]
akka.cluster.include-ref-node-in-replica-set = on
akka.actor.timeout = 30

View file

@ -2,7 +2,7 @@ akka.enabled-modules = ["cluster"]
akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1","node:node3"]
akka.cluster.include-ref-node-in-replica-set = on
akka.actor.timeout = 30

View file

@ -2,7 +2,7 @@ akka.enabled-modules = ["cluster"]
akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1","node:node3"]
akka.cluster.include-ref-node-in-replica-set = on
akka.actor.timeout = 30

View file

@ -2,7 +2,7 @@ akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-node1.router = "round-robin"
akka.actor.deployment.service-node1.cluster.preferred-nodes = ["node:node1"]
akka.actor.deployment.service-node1.cluster.replication-factor = 1
akka.actor.deployment.service-node1.nr-of-instances = 1
akka.actor.deployment.service-node2.router = "round-robin"
akka.actor.deployment.service-node2.cluster.preferred-nodes = ["node:node2"]
akka.actor.deployment.service-node2.cluster.replication-factor = 1
akka.actor.deployment.service-node2.nr-of-instances = 1

View file

@ -2,4 +2,4 @@ akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.preferred-nodes = ["node:node1"]
akka.actor.deployment.service-hello.cluster.replication-factor = 1
akka.actor.deployment.service-hello.nr-of-instances = 1

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 1
akka.actor.deployment.service-hello.nr-of-instances = 1

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 3
akka.actor.deployment.service-hello.nr-of-instances = 3

View file

@ -1,4 +1,4 @@
akka.enabled-modules = ["cluster"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "round-robin"
akka.actor.deployment.service-hello.cluster.replication-factor = 3
akka.actor.deployment.service-hello.nr-of-instances = 3

View file

@ -2,5 +2,5 @@ akka.enabled-modules = ["cluster"]
akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "akka.routing.ScatterGatherFirstCompletedRouter"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.timeout = 30

View file

@ -2,5 +2,5 @@ akka.enabled-modules = ["cluster"]
akka.event-handlers = ["akka.testkit.TestEventListener"]
akka.event-handler-level = "WARNING"
akka.actor.deployment.service-hello.router = "akka.routing.ScatterGatherFirstCompletedRouter"
akka.actor.deployment.service-hello.cluster.replication-factor = 2
akka.actor.deployment.service-hello.nr-of-instances = 2
akka.actor.timeout = 30

View file

@ -296,7 +296,7 @@ are:
- ``remove`` -- removes the actor from the clustered actor registry
The ``store`` method also allows you to specify a replication factor. The
``replicationFactor`` defines the number of (randomly picked) nodes in the cluster that
``nrOfInstances`` defines the number of (randomly picked) nodes in the cluster that
the stored actor should be automatically deployed to and instantiated locally on (using
``use``). If you leave this argument out then a replication factor of ``0`` will be used
which means that the actor will only be stored in the clustered actor registry and not
@ -310,11 +310,11 @@ on your use-case. Default is ``false``
This is the signatures for the ``store`` method (all different permutations of these methods are available for using from Java)::
def store[T <: Actor]
(actorRef: ActorRef, replicationFactor: Int = 0, serializeMailbox: Boolean = false)
(actorRef: ActorRef, nrOfInstances: Int = 0, serializeMailbox: Boolean = false)
(implicit format: Format[T]): ClusterNode
def store[T <: Actor]
(actorClass: Class[T], replicationFactor: Int = 0, serializeMailbox: Boolean = false)
(actorClass: Class[T], nrOfInstances: Int = 0, serializeMailbox: Boolean = false)
(implicit format: Format[T]): ClusterNode
The ``implicit format: Format[T]`` might look scary but this argument is chosen for you and passed in automatically by the compiler as long as you have imported the serialization typeclass for the actor you are storing, e.g. the ``HelloActorFormat`` (defined above and imported in the sample below).
@ -331,9 +331,9 @@ created actor::
val hello = actorOf[HelloActor].start.asInstanceOf[LocalActorRef]
val serializeMailbox = false
val replicationFactor = 5
val nrOfInstances = 5
clusterNode store (hello, serializeMailbox, replicationFactor)
clusterNode store (hello, serializeMailbox, nrOfInstances)
Here is an example of how to use ``store`` to cluster an actor by type::
@ -444,7 +444,7 @@ The workhorse for this is the ``send`` method (in different variations). The
``send`` methods take the following parameters:
- ``f`` -- the function you want to be invoked on the remote nodes in the cluster
- ``arg`` -- the argument to the function (not all of them have this parameter)
- ``replicationFactor`` -- the replication factor defining the number of nodes you want the function to be sent and invoked on
- ``nrOfInstances`` -- the replication factor defining the number of nodes you want the function to be sent and invoked on
You can currently send these function types to the cluster:
- ``Function0[Unit]`` -- takes no arguments and returns nothing

View file

@ -70,10 +70,9 @@ akka {
# default is "direct";
# if 'replication' is used then the only available router is "direct"
replication-factor = 3 # number of actor instances in the cluster
# available: positive integer (0-N) or the string "auto" for auto-scaling
# if "auto" is used then 'home' has no meaning
# default is '0', meaning no replicas;
nr-of-instances = 3 # number of actor instances in the cluster
# available: positive integer (1-N) or the string "auto" for auto-scaling
# default is '1'
# if the "direct" router is used then this element is ignored (always '1')
failure-detector { # failure detection scheme to use