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

format source with scalafmt, #26511

Browse source
This commit is contained in:
Auto Format 2019-03-13 10:56:20 +01:00 • committed by Patrik Nordwall
parent 2ba9b988df
commit 75579bed17
779 changed files with 15729 additions and 13096 deletions
Download patch file Download diff file Expand all files Collapse all files

23
akka-distributed-data/src/main/scala/akka/cluster/ddata/DurableStore.scala
View file

@ -103,10 +103,11 @@ object LmdbDurableStore {
private case object WriteBehind extends DeadLetterSuppression
private final case class Lmdb(env: Env[ByteBuffer],
db: Dbi[ByteBuffer],
keyBuffer: ByteBuffer,
valueBuffer: ByteBuffer)
private final case class Lmdb(
env: Env[ByteBuffer],
db: Dbi[ByteBuffer],
keyBuffer: ByteBuffer,
valueBuffer: ByteBuffer)
}
final class LmdbDurableStore(config: Config) extends Actor with ActorLogging {
@ -150,9 +151,10 @@ final class LmdbDurableStore(config: Config) extends Actor with ActorLogging {
val valueBuffer = ByteBuffer.allocateDirect(100 * 1024) // will grow when needed
if (log.isDebugEnabled)
log.debug("Init of LMDB in directory [{}] took [{} ms]",
dir.getCanonicalPath,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime - t0))
log.debug(
"Init of LMDB in directory [{}] took [{} ms]",
dir.getCanonicalPath,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime - t0))
val l = Lmdb(env, db, keyBuffer, valueBuffer)
_lmdb = OptionVal.Some(l)
l
@ -293,9 +295,10 @@ final class LmdbDurableStore(config: Config) extends Actor with ActorLogging {
}
tx.commit()
if (log.isDebugEnabled)
log.debug("store and commit of [{}] entries took [{} ms]",
pending.size,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime - t0))
log.debug(
"store and commit of [{}] entries took [{} ms]",
pending.size,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime - t0))
} catch {
case NonFatal(e) =>
import scala.collection.JavaConverters._

5
akka-distributed-data/src/main/scala/akka/cluster/ddata/GCounter.scala
View file

@ -41,8 +41,9 @@ object GCounter {
* This class is immutable, i.e. "modifying" methods return a new instance.
*/
@SerialVersionUID(1L)
final class GCounter private[akka] (private[akka] val state: Map[UniqueAddress, BigInt] = Map.empty,
override val delta: Option[GCounter] = None)
final class GCounter private[akka] (
private[akka] val state: Map[UniqueAddress, BigInt] = Map.empty,
override val delta: Option[GCounter] = None)
extends DeltaReplicatedData
with ReplicatedDelta
with ReplicatedDataSerialization

35
akka-distributed-data/src/main/scala/akka/cluster/ddata/ORMap.scala
View file

@ -68,9 +68,10 @@ object ORMap {
// PutDeltaOp contains ORSet delta and full value
/** INTERNAL API */
@InternalApi private[akka] final case class PutDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp,
value: (A, B),
zeroTag: ZeroTag)
@InternalApi private[akka] final case class PutDeltaOp[A, B <: ReplicatedData](
underlying: ORSet.DeltaOp,
value: (A, B),
zeroTag: ZeroTag)
extends AtomicDeltaOp[A, B] {
override def merge(that: DeltaOp): DeltaOp = that match {
case put: PutDeltaOp[A, B] if this.value._1 == put.value._1 =>
@ -93,9 +94,10 @@ object ORMap {
// UpdateDeltaOp contains ORSet delta and either delta of value (in case where underlying type supports deltas) or full value
/** INTERNAL API */
@InternalApi private[akka] final case class UpdateDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp,
values: Map[A, B],
zeroTag: ZeroTag)
@InternalApi private[akka] final case class UpdateDeltaOp[A, B <: ReplicatedData](
underlying: ORSet.DeltaOp,
values: Map[A, B],
zeroTag: ZeroTag)
extends AtomicDeltaOp[A, B] {
override def merge(that: DeltaOp): DeltaOp = that match {
case update: UpdateDeltaOp[A, B] =>
@ -117,15 +119,17 @@ object ORMap {
// RemoveDeltaOp does not contain any value at all - the propagated 'value' map would be empty
/** INTERNAL API */
@InternalApi private[akka] final case class RemoveDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp,
zeroTag: ZeroTag)
@InternalApi private[akka] final case class RemoveDeltaOp[A, B <: ReplicatedData](
underlying: ORSet.DeltaOp,
zeroTag: ZeroTag)
extends AtomicDeltaOp[A, B]
// RemoveKeyDeltaOp contains a single value - to provide the recipient with the removed key for value map
/** INTERNAL API */
@InternalApi private[akka] final case class RemoveKeyDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp,
removedKey: A,
zeroTag: ZeroTag)
@InternalApi private[akka] final case class RemoveKeyDeltaOp[A, B <: ReplicatedData](
underlying: ORSet.DeltaOp,
removedKey: A,
zeroTag: ZeroTag)
extends AtomicDeltaOp[A, B]
// DeltaGroup is effectively a causally ordered list of individual deltas
@ -169,10 +173,11 @@ object ORMap {
* This class is immutable, i.e. "modifying" methods return a new instance.
*/
@SerialVersionUID(1L)
final class ORMap[A, B <: ReplicatedData] private[akka] (private[akka] val keys: ORSet[A],
private[akka] val values: Map[A, B],
private[akka] val zeroTag: ZeroTag,
override val delta: Option[ORMap.DeltaOp] = None)
final class ORMap[A, B <: ReplicatedData] private[akka] (
private[akka] val keys: ORSet[A],
private[akka] val values: Map[A, B],
private[akka] val zeroTag: ZeroTag,
override val delta: Option[ORMap.DeltaOp] = None)
extends DeltaReplicatedData
with ReplicatedDataSerialization
with RemovedNodePruning {

24
akka-distributed-data/src/main/scala/akka/cluster/ddata/ORMultiMap.scala
View file

@ -65,8 +65,9 @@ object ORMultiMap {
* Note that on concurrent adds and removals for the same key (on the same set), removals can be lost.
*/
@SerialVersionUID(1L)
final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[A, ORSet[B]],
private[akka] val withValueDeltas: Boolean)
final class ORMultiMap[A, B] private[akka] (
private[akka] val underlying: ORMap[A, ORSet[B]],
private[akka] val withValueDeltas: Boolean)
extends DeltaReplicatedData
with ReplicatedDataSerialization
with RemovedNodePruning {
@ -82,8 +83,9 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
val newValues = newUnderlying.values.filterNot {
case (key, value) => !newUnderlying.keys.contains(key) && value.isEmpty
}
new ORMultiMap[A, B](new ORMap(newUnderlying.keys, newValues, newUnderlying.zeroTag, newUnderlying.delta),
withValueDeltas)
new ORMultiMap[A, B](
new ORMap(newUnderlying.keys, newValues, newUnderlying.zeroTag, newUnderlying.delta),
withValueDeltas)
} else
new ORMultiMap(underlying.merge(that.underlying), withValueDeltas)
} else throw new IllegalArgumentException("Trying to merge two ORMultiMaps of different map sub-type")
@ -305,10 +307,11 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
/**
* INTERNAL API
*/
@InternalApi private[akka] def replaceBinding(node: UniqueAddress,
key: A,
oldElement: B,
newElement: B): ORMultiMap[A, B] =
@InternalApi private[akka] def replaceBinding(
node: UniqueAddress,
key: A,
oldElement: B,
newElement: B): ORMultiMap[A, B] =
if (newElement != oldElement)
addBinding(node, key, newElement).removeBinding(node, key, oldElement)
else
@ -326,8 +329,9 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
val newValues = newUnderlying.values.filterNot {
case (key, value) => !newUnderlying.keys.contains(key) && value.isEmpty
}
new ORMultiMap[A, B](new ORMap(newUnderlying.keys, newValues, newUnderlying.zeroTag, newUnderlying.delta),
withValueDeltas)
new ORMultiMap[A, B](
new ORMap(newUnderlying.keys, newValues, newUnderlying.zeroTag, newUnderlying.delta),
withValueDeltas)
} else
new ORMultiMap(underlying.mergeDelta(thatDelta), withValueDeltas)

44
akka-distributed-data/src/main/scala/akka/cluster/ddata/ORSet.scala
View file

@ -126,8 +126,9 @@ object ORSet {
*/
@InternalApi private[akka] def subtractDots(dot: Dot, vvector: VersionVector): Dot = {
@tailrec def dropDots(remaining: List[(UniqueAddress, Long)],
acc: List[(UniqueAddress, Long)]): List[(UniqueAddress, Long)] =
@tailrec def dropDots(
remaining: List[(UniqueAddress, Long)],
acc: List[(UniqueAddress, Long)]): List[(UniqueAddress, Long)] =
remaining match {
case Nil => acc
case (d @ (node, v1)) :: rest =>
@ -160,9 +161,10 @@ object ORSet {
* INTERNAL API
* @see [[ORSet#merge]]
*/
@InternalApi private[akka] def mergeCommonKeys[A](commonKeys: Set[A],
lhs: ORSet[A],
rhs: ORSet[A]): Map[A, ORSet.Dot] =
@InternalApi private[akka] def mergeCommonKeys[A](
commonKeys: Set[A],
lhs: ORSet[A],
rhs: ORSet[A]): Map[A, ORSet.Dot] =
mergeCommonKeys(commonKeys.iterator, lhs, rhs)
private def mergeCommonKeys[A](commonKeys: Iterator[A], lhs: ORSet[A], rhs: ORSet[A]): Map[A, ORSet.Dot] = {
@ -231,16 +233,18 @@ object ORSet {
* INTERNAL API
* @see [[ORSet#merge]]
*/
@InternalApi private[akka] def mergeDisjointKeys[A](keys: Set[A],
elementsMap: Map[A, ORSet.Dot],
vvector: VersionVector,
accumulator: Map[A, ORSet.Dot]): Map[A, ORSet.Dot] =
@InternalApi private[akka] def mergeDisjointKeys[A](
keys: Set[A],
elementsMap: Map[A, ORSet.Dot],
vvector: VersionVector,
accumulator: Map[A, ORSet.Dot]): Map[A, ORSet.Dot] =
mergeDisjointKeys(keys.iterator, elementsMap, vvector, accumulator)
private def mergeDisjointKeys[A](keys: Iterator[A],
elementsMap: Map[A, ORSet.Dot],
vvector: VersionVector,
accumulator: Map[A, ORSet.Dot]): Map[A, ORSet.Dot] = {
private def mergeDisjointKeys[A](
keys: Iterator[A],
elementsMap: Map[A, ORSet.Dot],
vvector: VersionVector,
accumulator: Map[A, ORSet.Dot]): Map[A, ORSet.Dot] = {
keys.foldLeft(accumulator) {
case (acc, k) =>
val dots = elementsMap(k)
@ -285,9 +289,10 @@ object ORSet {
* This class is immutable, i.e. "modifying" methods return a new instance.
*/
@SerialVersionUID(1L)
final class ORSet[A] private[akka] (private[akka] val elementsMap: Map[A, ORSet.Dot],
private[akka] val vvector: VersionVector,
override val delta: Option[ORSet.DeltaOp] = None)
final class ORSet[A] private[akka] (
private[akka] val elementsMap: Map[A, ORSet.Dot],
private[akka] val vvector: VersionVector,
override val delta: Option[ORSet.DeltaOp] = None)
extends DeltaReplicatedData
with ReplicatedDataSerialization
with RemovedNodePruning
@ -530,9 +535,10 @@ final class ORSet[A] private[akka] (private[akka] val elementsMap: Map[A, ORSet.
new ORSet(updated, vvector.pruningCleanup(removedNode))
}
private def copy(elementsMap: Map[A, ORSet.Dot] = this.elementsMap,
vvector: VersionVector = this.vvector,
delta: Option[ORSet.DeltaOp] = this.delta): ORSet[A] =
private def copy(
elementsMap: Map[A, ORSet.Dot] = this.elementsMap,
vvector: VersionVector = this.vvector,
delta: Option[ORSet.DeltaOp] = this.delta): ORSet[A] =
new ORSet(elementsMap, vvector, delta)
// this class cannot be a `case class` because we need different `unapply`

5
akka-distributed-data/src/main/scala/akka/cluster/ddata/PNCounter.scala
View file

@ -205,8 +205,9 @@ final class PNCounter private[akka] (private[akka] val increments: GCounter, pri
increments.needPruningFrom(removedNode) || decrements.needPruningFrom(removedNode)
override def prune(removedNode: UniqueAddress, collapseInto: UniqueAddress): PNCounter =
copy(increments = increments.prune(removedNode, collapseInto),
decrements = decrements.prune(removedNode, collapseInto))
copy(
increments = increments.prune(removedNode, collapseInto),
decrements = decrements.prune(removedNode, collapseInto))
override def pruningCleanup(removedNode: UniqueAddress): PNCounter =
copy(increments = increments.pruningCleanup(removedNode), decrements = decrements.pruningCleanup(removedNode))

550
akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala
View file

@ -138,120 +138,129 @@ object ReplicatorSettings {
* `*` at the end of a key. All entries can be made durable by including "*"
* in the `Set`.
*/
final class ReplicatorSettings(val roles: Set[String],
val gossipInterval: FiniteDuration,
val notifySubscribersInterval: FiniteDuration,
val maxDeltaElements: Int,
val dispatcher: String,
val pruningInterval: FiniteDuration,
val maxPruningDissemination: FiniteDuration,
val durableStoreProps: Either[(String, Config), Props],
val durableKeys: Set[KeyId],
val pruningMarkerTimeToLive: FiniteDuration,
val durablePruningMarkerTimeToLive: FiniteDuration,
val deltaCrdtEnabled: Boolean,
val maxDeltaSize: Int) {
final class ReplicatorSettings(
val roles: Set[String],
val gossipInterval: FiniteDuration,
val notifySubscribersInterval: FiniteDuration,
val maxDeltaElements: Int,
val dispatcher: String,
val pruningInterval: FiniteDuration,
val maxPruningDissemination: FiniteDuration,
val durableStoreProps: Either[(String, Config), Props],
val durableKeys: Set[KeyId],
val pruningMarkerTimeToLive: FiniteDuration,
val durablePruningMarkerTimeToLive: FiniteDuration,
val deltaCrdtEnabled: Boolean,
val maxDeltaSize: Int) {
// for backwards compatibility
def this(role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration,
durableStoreProps: Either[(String, Config), Props],
durableKeys: Set[KeyId],
pruningMarkerTimeToLive: FiniteDuration,
durablePruningMarkerTimeToLive: FiniteDuration,
deltaCrdtEnabled: Boolean,
maxDeltaSize: Int) =
this(role.toSet,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive,
deltaCrdtEnabled,
maxDeltaSize)
def this(
role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration,
durableStoreProps: Either[(String, Config), Props],
durableKeys: Set[KeyId],
pruningMarkerTimeToLive: FiniteDuration,
durablePruningMarkerTimeToLive: FiniteDuration,
deltaCrdtEnabled: Boolean,
maxDeltaSize: Int) =
this(
role.toSet,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive,
deltaCrdtEnabled,
maxDeltaSize)
// For backwards compatibility
def this(role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration) =
this(roles = role.toSet,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
Right(Props.empty),
Set.empty,
6.hours,
10.days,
true,
200)
def this(
role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration) =
this(
roles = role.toSet,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
Right(Props.empty),
Set.empty,
6.hours,
10.days,
true,
200)
// For backwards compatibility
def this(role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration,
durableStoreProps: Either[(String, Config), Props],
durableKeys: Set[String]) =
this(role,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
6.hours,
10.days,
true,
200)
def this(
role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration,
durableStoreProps: Either[(String, Config), Props],
durableKeys: Set[String]) =
this(
role,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
6.hours,
10.days,
true,
200)
// For backwards compatibility
def this(role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration,
durableStoreProps: Either[(String, Config), Props],
durableKeys: Set[String],
pruningMarkerTimeToLive: FiniteDuration,
durablePruningMarkerTimeToLive: FiniteDuration,
deltaCrdtEnabled: Boolean) =
this(role,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive,
deltaCrdtEnabled,
200)
def this(
role: Option[String],
gossipInterval: FiniteDuration,
notifySubscribersInterval: FiniteDuration,
maxDeltaElements: Int,
dispatcher: String,
pruningInterval: FiniteDuration,
maxPruningDissemination: FiniteDuration,
durableStoreProps: Either[(String, Config), Props],
durableKeys: Set[String],
pruningMarkerTimeToLive: FiniteDuration,
durablePruningMarkerTimeToLive: FiniteDuration,
deltaCrdtEnabled: Boolean) =
this(
role,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive,
deltaCrdtEnabled,
200)
def withRole(role: String): ReplicatorSettings = copy(roles = ReplicatorSettings.roleOption(role).toSet)
@ -288,10 +297,12 @@ final class ReplicatorSettings(val roles: Set[String],
def withPruning(pruningInterval: FiniteDuration, maxPruningDissemination: FiniteDuration): ReplicatorSettings =
copy(pruningInterval = pruningInterval, maxPruningDissemination = maxPruningDissemination)
def withPruningMarkerTimeToLive(pruningMarkerTimeToLive: FiniteDuration,
durablePruningMarkerTimeToLive: FiniteDuration): ReplicatorSettings =
copy(pruningMarkerTimeToLive = pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive = durablePruningMarkerTimeToLive)
def withPruningMarkerTimeToLive(
pruningMarkerTimeToLive: FiniteDuration,
durablePruningMarkerTimeToLive: FiniteDuration): ReplicatorSettings =
copy(
pruningMarkerTimeToLive = pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive = durablePruningMarkerTimeToLive)
def withDurableStoreProps(durableStoreProps: Props): ReplicatorSettings =
copy(durableStoreProps = Right(durableStoreProps))
@ -316,32 +327,34 @@ final class ReplicatorSettings(val roles: Set[String],
def withMaxDeltaSize(maxDeltaSize: Int): ReplicatorSettings =
copy(maxDeltaSize = maxDeltaSize)
private def copy(roles: Set[String] = roles,
gossipInterval: FiniteDuration = gossipInterval,
notifySubscribersInterval: FiniteDuration = notifySubscribersInterval,
maxDeltaElements: Int = maxDeltaElements,
dispatcher: String = dispatcher,
pruningInterval: FiniteDuration = pruningInterval,
maxPruningDissemination: FiniteDuration = maxPruningDissemination,
durableStoreProps: Either[(String, Config), Props] = durableStoreProps,
durableKeys: Set[KeyId] = durableKeys,
pruningMarkerTimeToLive: FiniteDuration = pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive: FiniteDuration = durablePruningMarkerTimeToLive,
deltaCrdtEnabled: Boolean = deltaCrdtEnabled,
maxDeltaSize: Int = maxDeltaSize): ReplicatorSettings =
new ReplicatorSettings(roles,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive,
deltaCrdtEnabled,
maxDeltaSize)
private def copy(
roles: Set[String] = roles,
gossipInterval: FiniteDuration = gossipInterval,
notifySubscribersInterval: FiniteDuration = notifySubscribersInterval,
maxDeltaElements: Int = maxDeltaElements,
dispatcher: String = dispatcher,
pruningInterval: FiniteDuration = pruningInterval,
maxPruningDissemination: FiniteDuration = maxPruningDissemination,
durableStoreProps: Either[(String, Config), Props] = durableStoreProps,
durableKeys: Set[KeyId] = durableKeys,
pruningMarkerTimeToLive: FiniteDuration = pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive: FiniteDuration = durablePruningMarkerTimeToLive,
deltaCrdtEnabled: Boolean = deltaCrdtEnabled,
maxDeltaSize: Int = maxDeltaSize): ReplicatorSettings =
new ReplicatorSettings(
roles,
gossipInterval,
notifySubscribersInterval,
maxDeltaElements,
dispatcher,
pruningInterval,
maxPruningDissemination,
durableStoreProps,
durableKeys,
pruningMarkerTimeToLive,
durablePruningMarkerTimeToLive,
deltaCrdtEnabled,
maxDeltaSize)
}
object Replicator {
@ -350,8 +363,9 @@ object Replicator {
* Factory method for the [[akka.actor.Props]] of the [[Replicator]] actor.
*/
def props(settings: ReplicatorSettings): Props = {
require(settings.durableKeys.isEmpty || (settings.durableStoreProps != Right(Props.empty)),
"durableStoreProps must be defined when durableKeys are defined")
require(
settings.durableKeys.isEmpty || (settings.durableStoreProps != Right(Props.empty)),
"durableStoreProps must be defined when durableKeys are defined")
Props(new Replicator(settings)).withDeploy(Deploy.local).withDispatcher(settings.dispatcher)
}
@ -578,10 +592,11 @@ object Replicator {
* way to pass contextual information (e.g. original sender) without having to use `ask`
* or local correlation data structures.
*/
def apply[A <: ReplicatedData](key: Key[A],
initial: A,
writeConsistency: WriteConsistency,
request: Option[Any] = None)(modify: A => A): Update[A] =
def apply[A <: ReplicatedData](
key: Key[A],
initial: A,
writeConsistency: WriteConsistency,
request: Option[Any] = None)(modify: A => A): Update[A] =
Update(key, writeConsistency, request)(modifyWithInitial(initial, modify))
private def modifyWithInitial[A <: ReplicatedData](initial: A, modify: A => A): Option[A] => A = {
@ -632,11 +647,12 @@ object Replicator {
* way to pass contextual information (e.g. original sender) without having to use `ask`
* or local correlation data structures.
*/
def this(key: Key[A],
initial: A,
writeConsistency: WriteConsistency,
request: Optional[Any],
modify: JFunction[A, A]) =
def this(
key: Key[A],
initial: A,
writeConsistency: WriteConsistency,
request: Optional[Any],
modify: JFunction[A, A]) =
this(key, writeConsistency, Option(request.orElse(null)))(
Update.modifyWithInitial(initial, data => modify.apply(data)))
@ -669,10 +685,11 @@ object Replicator {
* If the `modify` function of the [[Update]] throws an exception the reply message
* will be this `ModifyFailure` message. The original exception is included as `cause`.
*/
final case class ModifyFailure[A <: ReplicatedData](key: Key[A],
errorMessage: String,
cause: Throwable,
request: Option[Any])
final case class ModifyFailure[A <: ReplicatedData](
key: Key[A],
errorMessage: String,
cause: Throwable,
request: Option[Any])
extends UpdateFailure[A] {
override def toString: String = s"ModifyFailure [$key]: $errorMessage"
}
@ -792,9 +809,10 @@ object Replicator {
/**
* The `DataEnvelope` wraps a data entry and carries state of the pruning process for the entry.
*/
final case class DataEnvelope(data: ReplicatedData,
pruning: Map[UniqueAddress, PruningState] = Map.empty,
deltaVersions: VersionVector = VersionVector.empty)
final case class DataEnvelope(
data: ReplicatedData,
pruning: Map[UniqueAddress, PruningState] = Map.empty,
deltaVersions: VersionVector = VersionVector.empty)
extends ReplicatorMessage {
import PruningState._
@ -817,8 +835,9 @@ object Replicator {
}
def initRemovedNodePruning(removed: UniqueAddress, owner: UniqueAddress): DataEnvelope = {
copy(pruning = pruning.updated(removed, PruningInitialized(owner, Set.empty)),
deltaVersions = cleanedDeltaVersions(removed))
copy(
pruning = pruning.updated(removed, PruningInitialized(owner, Set.empty)),
deltaVersions = cleanedDeltaVersions(removed))
}
def prune(from: UniqueAddress, pruningPerformed: PruningPerformed): DataEnvelope = {
@ -828,9 +847,10 @@ object Replicator {
pruning(from) match {
case PruningInitialized(owner, _) =>
val prunedData = dataWithRemovedNodePruning.prune(from, owner)
copy(data = prunedData,
pruning = pruning.updated(from, pruningPerformed),
deltaVersions = cleanedDeltaVersions(from))
copy(
data = prunedData,
pruning = pruning.updated(from, pruningPerformed),
deltaVersions = cleanedDeltaVersions(from))
case _ =>
this
}
@ -874,9 +894,10 @@ object Replicator {
val mergedDeltaVersions = cleanedDV.merge(cleanedOtherDV)
// cleanup both sides before merging, `merge(otherData: ReplicatedData)` will cleanup other.data
copy(data = cleaned(data, filteredMergedPruning),
deltaVersions = mergedDeltaVersions,
pruning = filteredMergedPruning).merge(other.data)
copy(
data = cleaned(data, filteredMergedPruning),
deltaVersions = mergedDeltaVersions,
pruning = filteredMergedPruning).merge(other.data)
}
def merge(otherData: ReplicatedData): DataEnvelope = {
@ -1166,8 +1187,9 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
val selfUniqueAddress = cluster.selfUniqueAddress
require(!cluster.isTerminated, "Cluster node must not be terminated")
require(roles.subsetOf(cluster.selfRoles),
s"This cluster member [${selfAddress}] doesn't have all the roles [${roles.mkString(", ")}]")
require(
roles.subsetOf(cluster.selfRoles),
s"This cluster member [${selfAddress}] doesn't have all the roles [${roles.mkString(", ")}]")
//Start periodic gossip to random nodes in cluster
import context.dispatcher
@ -1210,15 +1232,16 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
override def createDeltaPropagation(deltas: Map[KeyId, (ReplicatedData, Long, Long)]): DeltaPropagation = {
// Important to include the pruning state in the deltas. For example if the delta is based
// on an entry that has been pruned but that has not yet been performed on the target node.
DeltaPropagation(selfUniqueAddress,
reply = false,
deltas.iterator.collect {
case (key, (d, fromSeqNr, toSeqNr)) if d != NoDeltaPlaceholder =>
getData(key) match {
case Some(envelope) => key -> Delta(envelope.copy(data = d), fromSeqNr, toSeqNr)
case None => key -> Delta(DataEnvelope(d), fromSeqNr, toSeqNr)
}
}.toMap)
DeltaPropagation(
selfUniqueAddress,
reply = false,
deltas.iterator.collect {
case (key, (d, fromSeqNr, toSeqNr)) if d != NoDeltaPlaceholder =>
getData(key) match {
case Some(envelope) => key -> Delta(envelope.copy(data = d), fromSeqNr, toSeqNr)
case None => key -> Delta(DataEnvelope(d), fromSeqNr, toSeqNr)
}
}.toMap)
}
}
val deltaPropagationTask: Option[Cancellable] =
@ -1302,9 +1325,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def fromDurableStore: Boolean = sender() == durableStore && sender() != context.system.deadLetters
OneForOneStrategy()(({
case e @ (_: DurableStore.LoadFailed | _: ActorInitializationException) if fromDurableStore =>
log.error(e,
"Stopping distributed-data Replicator due to load or startup failure in durable store, caused by: {}",
if (e.getCause eq null) "" else e.getCause.getMessage)
log.error(
e,
"Stopping distributed-data Replicator due to load or startup failure in durable store, caused by: {}",
if (e.getCause eq null) "" else e.getCause.getMessage)
context.stop(self)
SupervisorStrategy.Stop
}: SupervisorStrategy.Decider).orElse(SupervisorStrategy.defaultDecider))
@ -1344,10 +1368,11 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
}
}
case LoadAllCompleted =>
log.debug("Loading {} entries from durable store took {} ms, stashed {}",
count,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime),
stash.size)
log.debug(
"Loading {} entries from durable store took {} ms, stashed {}",
count,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime),
stash.size)
context.become(normalReceive)
unstashAll()
self ! FlushChanges
@ -1429,10 +1454,11 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def isLocalSender(): Boolean = !replyTo.path.address.hasGlobalScope
def receiveUpdate(key: KeyR,
modify: Option[ReplicatedData] => ReplicatedData,
writeConsistency: WriteConsistency,
req: Option[Any]): Unit = {
def receiveUpdate(
key: KeyR,
modify: Option[ReplicatedData] => ReplicatedData,
writeConsistency: WriteConsistency,
req: Option[Any]): Unit = {
val localValue = getData(key.id)
def deltaOrPlaceholder(d: DeltaReplicatedData): Option[ReplicatedDelta] = {
@ -1476,9 +1502,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
val durable = isDurable(key.id)
if (isLocalUpdate(writeConsistency)) {
if (durable)
durableStore ! Store(key.id,
new DurableDataEnvelope(newEnvelope),
Some(StoreReply(UpdateSuccess(key, req), StoreFailure(key, req), replyTo)))
durableStore ! Store(
key.id,
new DurableDataEnvelope(newEnvelope),
Some(StoreReply(UpdateSuccess(key, req), StoreFailure(key, req), replyTo)))
else
replyTo ! UpdateSuccess(key, req)
} else {
@ -1496,9 +1523,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
.props(key, writeEnvelope, writeDelta, writeConsistency, req, nodes, unreachable, replyTo, durable)
.withDispatcher(context.props.dispatcher))
if (durable) {
durableStore ! Store(key.id,
new DurableDataEnvelope(newEnvelope),
Some(StoreReply(UpdateSuccess(key, req), StoreFailure(key, req), writeAggregator)))
durableStore ! Store(
key.id,
new DurableDataEnvelope(newEnvelope),
Some(StoreReply(UpdateSuccess(key, req), StoreFailure(key, req), writeAggregator)))
}
}
case Failure(e: DataDeleted[_]) =>
@ -1591,9 +1619,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
val durable = isDurable(key.id)
if (isLocalUpdate(consistency)) {
if (durable)
durableStore ! Store(key.id,
new DurableDataEnvelope(DeletedEnvelope),
Some(StoreReply(DeleteSuccess(key, req), StoreFailure(key, req), replyTo)))
durableStore ! Store(
key.id,
new DurableDataEnvelope(DeletedEnvelope),
Some(StoreReply(DeleteSuccess(key, req), StoreFailure(key, req), replyTo)))
else
replyTo ! DeleteSuccess(key, req)
} else {
@ -1603,9 +1632,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
.props(key, DeletedEnvelope, None, consistency, req, nodes, unreachable, replyTo, durable)
.withDispatcher(context.props.dispatcher))
if (durable) {
durableStore ! Store(key.id,
new DurableDataEnvelope(DeletedEnvelope),
Some(StoreReply(DeleteSuccess(key, req), StoreFailure(key, req), writeAggregator)))
durableStore ! Store(
key.id,
new DurableDataEnvelope(DeletedEnvelope),
Some(StoreReply(DeleteSuccess(key, req), StoreFailure(key, req), writeAggregator)))
}
}
}
@ -1719,11 +1749,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
try {
val isDebugEnabled = log.isDebugEnabled
if (isDebugEnabled)
log.debug("Received DeltaPropagation from [{}], containing [{}]",
fromNode.address,
deltas
.collect { case (key, Delta(_, fromSeqNr, toSeqNr)) => s"$key $fromSeqNr-$toSeqNr" }
.mkString(", "))
log.debug(
"Received DeltaPropagation from [{}], containing [{}]",
fromNode.address,
deltas.collect { case (key, Delta(_, fromSeqNr, toSeqNr)) => s"$key $fromSeqNr-$toSeqNr" }.mkString(", "))
if (isNodeRemoved(fromNode, deltas.keys)) {
// Late message from a removed node.
@ -1736,27 +1765,30 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
val currentSeqNr = getDeltaSeqNr(key, fromNode)
if (currentSeqNr >= toSeqNr) {
if (isDebugEnabled)
log.debug("Skipping DeltaPropagation from [{}] for [{}] because toSeqNr [{}] already handled [{}]",
fromNode.address,
key,
toSeqNr,
currentSeqNr)
log.debug(
"Skipping DeltaPropagation from [{}] for [{}] because toSeqNr [{}] already handled [{}]",
fromNode.address,
key,
toSeqNr,
currentSeqNr)
if (reply) replyTo ! WriteAck
} else if (fromSeqNr > (currentSeqNr + 1)) {
if (isDebugEnabled)
log.debug("Skipping DeltaPropagation from [{}] for [{}] because missing deltas between [{}-{}]",
fromNode.address,
key,
currentSeqNr + 1,
fromSeqNr - 1)
log.debug(
"Skipping DeltaPropagation from [{}] for [{}] because missing deltas between [{}-{}]",
fromNode.address,
key,
currentSeqNr + 1,
fromSeqNr - 1)
if (reply) replyTo ! DeltaNack
} else {
if (isDebugEnabled)
log.debug("Applying DeltaPropagation from [{}] for [{}] with sequence numbers [{}], current was [{}]",
fromNode.address,
key,
s"$fromSeqNr-$toSeqNr",
currentSeqNr)
log.debug(
"Applying DeltaPropagation from [{}] for [{}] with sequence numbers [{}], current was [{}]",
fromNode.address,
key,
s"$fromSeqNr-$toSeqNr",
currentSeqNr)
val newEnvelope = envelope.copy(deltaVersions = VersionVector(fromNode, toSeqNr))
writeAndStore(key, newEnvelope, reply)
}
@ -1812,11 +1844,12 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def receiveStatus(otherDigests: Map[KeyId, Digest], chunk: Int, totChunks: Int): Unit = {
if (log.isDebugEnabled)
log.debug("Received gossip status from [{}], chunk [{}] of [{}] containing [{}]",
replyTo.path.address,
(chunk + 1),
totChunks,
otherDigests.keys.mkString(", "))
log.debug(
"Received gossip status from [{}], chunk [{}] of [{}] containing [{}]",
replyTo.path.address,
(chunk + 1),
totChunks,
otherDigests.keys.mkString(", "))
def isOtherDifferent(key: KeyId, otherDigest: Digest): Boolean = {
val d = getDigest(key)
@ -1840,9 +1873,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
val myMissingKeys = otherKeys.diff(myKeys)
if (myMissingKeys.nonEmpty) {
if (log.isDebugEnabled)
log.debug("Sending gossip status to [{}], requesting missing [{}]",
replyTo.path.address,
myMissingKeys.mkString(", "))
log.debug(
"Sending gossip status to [{}], requesting missing [{}]",
replyTo.path.address,
myMissingKeys.mkString(", "))
val status = Status(myMissingKeys.iterator.map(k => k -> NotFoundDigest).toMap, chunk, totChunks)
replyTo ! status
}
@ -2131,15 +2165,16 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
* INTERNAL API
*/
@InternalApi private[akka] object WriteAggregator {
def props(key: KeyR,
envelope: Replicator.Internal.DataEnvelope,
delta: Option[Replicator.Internal.Delta],
consistency: Replicator.WriteConsistency,
req: Option[Any],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean): Props =
def props(
key: KeyR,
envelope: Replicator.Internal.DataEnvelope,
delta: Option[Replicator.Internal.Delta],
consistency: Replicator.WriteConsistency,
req: Option[Any],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean): Props =
Props(new WriteAggregator(key, envelope, delta, consistency, req, nodes, unreachable, replyTo, durable))
.withDeploy(Deploy.local)
}
@ -2147,15 +2182,16 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
/**
* INTERNAL API
*/
@InternalApi private[akka] class WriteAggregator(key: KeyR,
envelope: Replicator.Internal.DataEnvelope,
delta: Option[Replicator.Internal.Delta],
consistency: Replicator.WriteConsistency,
req: Option[Any],
override val nodes: Set[Address],
override val unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean)
@InternalApi private[akka] class WriteAggregator(
key: KeyR,
envelope: Replicator.Internal.DataEnvelope,
delta: Option[Replicator.Internal.Delta],
consistency: Replicator.WriteConsistency,
req: Option[Any],
override val nodes: Set[Address],
override val unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean)
extends ReadWriteAggregator {
import Replicator._
@ -2270,13 +2306,14 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
* INTERNAL API
*/
@InternalApi private[akka] object ReadAggregator {
def props(key: KeyR,
consistency: Replicator.ReadConsistency,
req: Option[Any],
nodes: Set[Address],
unreachable: Set[Address],
localValue: Option[Replicator.Internal.DataEnvelope],
replyTo: ActorRef): Props =
def props(
key: KeyR,
consistency: Replicator.ReadConsistency,
req: Option[Any],
nodes: Set[Address],
unreachable: Set[Address],
localValue: Option[Replicator.Internal.DataEnvelope],
replyTo: ActorRef): Props =
Props(new ReadAggregator(key, consistency, req, nodes, unreachable, localValue, replyTo)).withDeploy(Deploy.local)
}
@ -2284,13 +2321,14 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
/**
* INTERNAL API
*/
@InternalApi private[akka] class ReadAggregator(key: KeyR,
consistency: Replicator.ReadConsistency,
req: Option[Any],
override val nodes: Set[Address],
override val unreachable: Set[Address],
localValue: Option[Replicator.Internal.DataEnvelope],
replyTo: ActorRef)
@InternalApi private[akka] class ReadAggregator(
key: KeyR,
consistency: Replicator.ReadConsistency,
req: Option[Any],
override val nodes: Set[Address],
override val unreachable: Set[Address],
localValue: Option[Replicator.Internal.DataEnvelope],
replyTo: ActorRef)
extends ReadWriteAggregator {
import Replicator._

7
akka-distributed-data/src/main/scala/akka/cluster/ddata/VersionVector.scala
View file

@ -182,9 +182,10 @@ sealed abstract class VersionVector extends ReplicatedData with ReplicatedDataSe
private final def compareOnlyTo(that: VersionVector, order: Ordering): Ordering = {
def nextOrElse[A](iter: Iterator[A], default: A): A = if (iter.hasNext) iter.next() else default
def compare(i1: Iterator[(UniqueAddress, Long)],
i2: Iterator[(UniqueAddress, Long)],
requestedOrder: Ordering): Ordering = {
def compare(
i1: Iterator[(UniqueAddress, Long)],
i2: Iterator[(UniqueAddress, Long)],
requestedOrder: Ordering): Ordering = {
@tailrec
def compareNext(nt1: (UniqueAddress, Long), nt2: (UniqueAddress, Long), currentOrder: Ordering): Ordering =
if ((requestedOrder ne FullOrder) && (currentOrder ne Same) && (currentOrder ne requestedOrder)) currentOrder

180
akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatedDataSerializer.scala
View file

@ -115,9 +115,10 @@ private object ReplicatedDataSerializer {
builder.setLongKey(key).setValue(value).build()
override def setIntKey(builder: rd.ORMap.Entry.Builder, key: Int, value: dm.OtherMessage): rd.ORMap.Entry =
builder.setIntKey(key).setValue(value).build()
override def setOtherKey(builder: rd.ORMap.Entry.Builder,
key: dm.OtherMessage,
value: dm.OtherMessage): rd.ORMap.Entry = builder.setOtherKey(key).setValue(value).build()
override def setOtherKey(
builder: rd.ORMap.Entry.Builder,
key: dm.OtherMessage,
value: dm.OtherMessage): rd.ORMap.Entry = builder.setOtherKey(key).setValue(value).build()
override def hasStringKey(entry: rd.ORMap.Entry): Boolean = entry.hasStringKey
override def getStringKey(entry: rd.ORMap.Entry): String = entry.getStringKey
override def hasIntKey(entry: rd.ORMap.Entry): Boolean = entry.hasIntKey
@ -138,9 +139,10 @@ private object ReplicatedDataSerializer {
builder.setLongKey(key).setValue(value).build()
override def setIntKey(builder: rd.LWWMap.Entry.Builder, key: Int, value: rd.LWWRegister): rd.LWWMap.Entry =
builder.setIntKey(key).setValue(value).build()
override def setOtherKey(builder: rd.LWWMap.Entry.Builder,
key: OtherMessage,
value: rd.LWWRegister): rd.LWWMap.Entry = builder.setOtherKey(key).setValue(value).build()
override def setOtherKey(
builder: rd.LWWMap.Entry.Builder,
key: OtherMessage,
value: rd.LWWRegister): rd.LWWMap.Entry = builder.setOtherKey(key).setValue(value).build()
override def hasStringKey(entry: rd.LWWMap.Entry): Boolean = entry.hasStringKey
override def getStringKey(entry: rd.LWWMap.Entry): String = entry.getStringKey
override def hasIntKey(entry: rd.LWWMap.Entry): Boolean = entry.hasIntKey
@ -155,20 +157,24 @@ private object ReplicatedDataSerializer {
implicit object PNCounterMapEntry
extends ProtoMapEntryWriter[rd.PNCounterMap.Entry, rd.PNCounterMap.Entry.Builder, rd.PNCounter]
with ProtoMapEntryReader[rd.PNCounterMap.Entry, rd.PNCounter] {
override def setStringKey(builder: rd.PNCounterMap.Entry.Builder,
key: String,
value: rd.PNCounter): rd.PNCounterMap.Entry =
override def setStringKey(
builder: rd.PNCounterMap.Entry.Builder,
key: String,
value: rd.PNCounter): rd.PNCounterMap.Entry =
builder.setStringKey(key).setValue(value).build()
override def setLongKey(builder: rd.PNCounterMap.Entry.Builder,
key: Long,
value: rd.PNCounter): rd.PNCounterMap.Entry =
override def setLongKey(
builder: rd.PNCounterMap.Entry.Builder,
key: Long,
value: rd.PNCounter): rd.PNCounterMap.Entry =
builder.setLongKey(key).setValue(value).build()
override def setIntKey(builder: rd.PNCounterMap.Entry.Builder,
key: Int,
value: rd.PNCounter): rd.PNCounterMap.Entry = builder.setIntKey(key).setValue(value).build()
override def setOtherKey(builder: rd.PNCounterMap.Entry.Builder,
key: OtherMessage,
value: rd.PNCounter): rd.PNCounterMap.Entry =
override def setIntKey(
builder: rd.PNCounterMap.Entry.Builder,
key: Int,
value: rd.PNCounter): rd.PNCounterMap.Entry = builder.setIntKey(key).setValue(value).build()
override def setOtherKey(
builder: rd.PNCounterMap.Entry.Builder,
key: OtherMessage,
value: rd.PNCounter): rd.PNCounterMap.Entry =
builder.setOtherKey(key).setValue(value).build()
override def hasStringKey(entry: rd.PNCounterMap.Entry): Boolean = entry.hasStringKey
override def getStringKey(entry: rd.PNCounterMap.Entry): String = entry.getStringKey
@ -190,9 +196,10 @@ private object ReplicatedDataSerializer {
builder.setLongKey(key).setValue(value).build()
override def setIntKey(builder: rd.ORMultiMap.Entry.Builder, key: Int, value: rd.ORSet): rd.ORMultiMap.Entry =
builder.setIntKey(key).setValue(value).build()
override def setOtherKey(builder: rd.ORMultiMap.Entry.Builder,
key: dm.OtherMessage,
value: rd.ORSet): rd.ORMultiMap.Entry = builder.setOtherKey(key).setValue(value).build()
override def setOtherKey(
builder: rd.ORMultiMap.Entry.Builder,
key: dm.OtherMessage,
value: rd.ORSet): rd.ORMultiMap.Entry = builder.setOtherKey(key).setValue(value).build()
override def hasStringKey(entry: rd.ORMultiMap.Entry): Boolean = entry.hasStringKey
override def getStringKey(entry: rd.ORMultiMap.Entry): String = entry.getStringKey
override def hasIntKey(entry: rd.ORMultiMap.Entry): Boolean = entry.hasIntKey
@ -207,21 +214,25 @@ private object ReplicatedDataSerializer {
implicit object ORMapDeltaGroupEntry
extends ProtoMapEntryWriter[rd.ORMapDeltaGroup.MapEntry, rd.ORMapDeltaGroup.MapEntry.Builder, dm.OtherMessage]
with ProtoMapEntryReader[rd.ORMapDeltaGroup.MapEntry, dm.OtherMessage] {
override def setStringKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: String,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
override def setStringKey(
builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: String,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
builder.setStringKey(key).setValue(value).build()
override def setLongKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: Long,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
override def setLongKey(
builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: Long,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
builder.setLongKey(key).setValue(value).build()
override def setIntKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: Int,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
override def setIntKey(
builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: Int,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
builder.setIntKey(key).setValue(value).build()
override def setOtherKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: dm.OtherMessage,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
override def setOtherKey(
builder: rd.ORMapDeltaGroup.MapEntry.Builder,
key: dm.OtherMessage,
value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry =
builder.setOtherKey(key).setValue(value).build()
override def hasStringKey(entry: rd.ORMapDeltaGroup.MapEntry): Boolean = entry.hasStringKey
override def getStringKey(entry: rd.ORMapDeltaGroup.MapEntry): String = entry.getStringKey
@ -590,9 +601,10 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
lwwRegisterFromProto(rd.LWWRegister.parseFrom(bytes))
def lwwRegisterFromProto(lwwRegister: rd.LWWRegister): LWWRegister[Any] =
new LWWRegister(uniqueAddressFromProto(lwwRegister.getNode),
otherMessageFromProto(lwwRegister.getState),
lwwRegister.getTimestamp)
new LWWRegister(
uniqueAddressFromProto(lwwRegister.getNode),
otherMessageFromProto(lwwRegister.getState),
lwwRegister.getTimestamp)
def gcounterToProto(gcounter: GCounter): rd.GCounter = {
val b = rd.GCounter.newBuilder()
@ -628,20 +640,23 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
pncounterFromProto(rd.PNCounter.parseFrom(bytes))
def pncounterFromProto(pncounter: rd.PNCounter): PNCounter = {
new PNCounter(increments = gcounterFromProto(pncounter.getIncrements),
decrements = gcounterFromProto(pncounter.getDecrements))
new PNCounter(
increments = gcounterFromProto(pncounter.getIncrements),
decrements = gcounterFromProto(pncounter.getDecrements))
}
/*
* Convert a Map[A, B] to an Iterable[Entry] where Entry is the protobuf map entry.
*/
private def getEntries[IKey,
IValue,
EntryBuilder <: GeneratedMessage.Builder[EntryBuilder],
PEntry <: GeneratedMessage,
PValue <: GeneratedMessage](input: Map[IKey, IValue],
createBuilder: () => EntryBuilder,
valueConverter: IValue => PValue)(
private def getEntries[
IKey,
IValue,
EntryBuilder <: GeneratedMessage.Builder[EntryBuilder],
PEntry <: GeneratedMessage,
PValue <: GeneratedMessage](
input: Map[IKey, IValue],
createBuilder: () => EntryBuilder,
valueConverter: IValue => PValue)(
implicit comparator: Comparator[PEntry],
eh: ProtoMapEntryWriter[PEntry, EntryBuilder, PValue]): java.lang.Iterable[PEntry] = {
// The resulting Iterable needs to be ordered deterministically in order to create same signature upon serializing same data
@ -681,8 +696,9 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
}
def ormapFromProto(ormap: rd.ORMap): ORMap[Any, ReplicatedData] = {
val entries = mapTypeFromProto(ormap.getEntriesList,
(v: dm.OtherMessage) => otherMessageFromProto(v).asInstanceOf[ReplicatedData])
val entries = mapTypeFromProto(
ormap.getEntriesList,
(v: dm.OtherMessage) => otherMessageFromProto(v).asInstanceOf[ReplicatedData])
new ORMap(keys = orsetFromProto(ormap.getKeys), entries, ORMap.VanillaORMapTag)
}
@ -766,25 +782,31 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
.map { entry =>
if (entry.getOperation == rd.ORMapDeltaOp.ORMapPut) {
val map =
singleMapEntryFromProto(entry.getEntryDataList,
(v: dm.OtherMessage) => otherMessageFromProto(v).asInstanceOf[ReplicatedData])
ORMap.PutDeltaOp(ORSet.AddDeltaOp(orsetFromProto(entry.getUnderlying)),
map.head,
zeroTagFromCode(entry.getZeroTag))
singleMapEntryFromProto(
entry.getEntryDataList,
(v: dm.OtherMessage) => otherMessageFromProto(v).asInstanceOf[ReplicatedData])
ORMap.PutDeltaOp(
ORSet.AddDeltaOp(orsetFromProto(entry.getUnderlying)),
map.head,
zeroTagFromCode(entry.getZeroTag))
} else if (entry.getOperation == rd.ORMapDeltaOp.ORMapRemove) {
ORMap.RemoveDeltaOp(ORSet.RemoveDeltaOp(orsetFromProto(entry.getUnderlying)),
zeroTagFromCode(entry.getZeroTag))
ORMap.RemoveDeltaOp(
ORSet.RemoveDeltaOp(orsetFromProto(entry.getUnderlying)),
zeroTagFromCode(entry.getZeroTag))
} else if (entry.getOperation == rd.ORMapDeltaOp.ORMapRemoveKey) {
val elem = singleKeyEntryFromProto(entry.getEntryDataList.asScala.headOption)
ORMap.RemoveKeyDeltaOp(ORSet.RemoveDeltaOp(orsetFromProto(entry.getUnderlying)),
elem,
zeroTagFromCode(entry.getZeroTag))
ORMap.RemoveKeyDeltaOp(
ORSet.RemoveDeltaOp(orsetFromProto(entry.getUnderlying)),
elem,
zeroTagFromCode(entry.getZeroTag))
} else if (entry.getOperation == rd.ORMapDeltaOp.ORMapUpdate) {
val map = mapTypeFromProto(entry.getEntryDataList,
(v: dm.OtherMessage) => otherMessageFromProto(v).asInstanceOf[ReplicatedDelta])
ORMap.UpdateDeltaOp(ORSet.AddDeltaOp(orsetFromProto(entry.getUnderlying)),
map,
zeroTagFromCode(entry.getZeroTag))
val map = mapTypeFromProto(
entry.getEntryDataList,
(v: dm.OtherMessage) => otherMessageFromProto(v).asInstanceOf[ReplicatedDelta])
ORMap.UpdateDeltaOp(
ORSet.AddDeltaOp(orsetFromProto(entry.getUnderlying)),
map,
zeroTagFromCode(entry.getZeroTag))
} else
throw new NotSerializableException(s"Unknown ORMap delta operation ${entry.getOperation}")
}
@ -862,16 +884,18 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
createEntry(rd.ORMapDeltaOp.ORMapPut, op.asInstanceOf[ORSet.AddDeltaOp[_]].underlying, Map(pair), zt.value))
case ORMap.RemoveDeltaOp(op, zt) =>
b.addEntries(
createEntry(rd.ORMapDeltaOp.ORMapRemove,
op.asInstanceOf[ORSet.RemoveDeltaOp[_]].underlying,
Map.empty,
zt.value))
createEntry(
rd.ORMapDeltaOp.ORMapRemove,
op.asInstanceOf[ORSet.RemoveDeltaOp[_]].underlying,
Map.empty,
zt.value))
case ORMap.RemoveKeyDeltaOp(op, k, zt) =>
b.addEntries(
createEntryWithKey(rd.ORMapDeltaOp.ORMapRemoveKey,
op.asInstanceOf[ORSet.RemoveDeltaOp[_]].underlying,
k,
zt.value))
createEntryWithKey(
rd.ORMapDeltaOp.ORMapRemoveKey,
op.asInstanceOf[ORSet.RemoveDeltaOp[_]].underlying,
k,
zt.value))
case ORMap.UpdateDeltaOp(op, m, zt) =>
b.addEntries(
createEntry(rd.ORMapDeltaOp.ORMapUpdate, op.asInstanceOf[ORSet.AddDeltaOp[_]].underlying, m, zt.value))
@ -930,13 +954,15 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
if (multimap.hasWithValueDeltas)
multimap.getWithValueDeltas
else false
new ORMultiMap(new ORMap(keys = orsetFromProto(multimap.getKeys),
entries,
if (withValueDeltas)
ORMultiMap.ORMultiMapWithValueDeltasTag
else
ORMultiMap.ORMultiMapTag),
withValueDeltas)
new ORMultiMap(
new ORMap(
keys = orsetFromProto(multimap.getKeys),
entries,
if (withValueDeltas)
ORMultiMap.ORMultiMapWithValueDeltasTag
else
ORMultiMap.ORMultiMapTag),
withValueDeltas)
}
def keyIdToBinary(id: String): Array[Byte] =

67
akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializer.scala
View file

@ -184,27 +184,25 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
val DeltaPropagationManifest = "Q"
val DeltaNackManifest = "R"
private val fromBinaryMap = collection.immutable.HashMap[String, Array[Byte] => AnyRef](GetManifest -> getFromBinary,
GetSuccessManifest -> getSuccessFromBinary,
NotFoundManifest -> notFoundFromBinary,
GetFailureManifest -> getFailureFromBinary,
SubscribeManifest -> subscribeFromBinary,
UnsubscribeManifest -> unsubscribeFromBinary,
ChangedManifest -> changedFromBinary,
DataEnvelopeManifest -> dataEnvelopeFromBinary,
WriteManifest -> writeFromBinary,
WriteAckManifest -> (_ =>
WriteAck),
ReadManifest -> readFromBinary,
ReadResultManifest -> readResultFromBinary,
StatusManifest -> statusFromBinary,
GossipManifest -> gossipFromBinary,
DeltaPropagationManifest -> deltaPropagationFromBinary,
WriteNackManifest -> (_ =>
WriteNack),
DeltaNackManifest -> (_ =>
DeltaNack),
DurableDataEnvelopeManifest -> durableDataEnvelopeFromBinary)
private val fromBinaryMap = collection.immutable.HashMap[String, Array[Byte] => AnyRef](
GetManifest -> getFromBinary,
GetSuccessManifest -> getSuccessFromBinary,
NotFoundManifest -> notFoundFromBinary,
GetFailureManifest -> getFailureFromBinary,
SubscribeManifest -> subscribeFromBinary,
UnsubscribeManifest -> unsubscribeFromBinary,
ChangedManifest -> changedFromBinary,
DataEnvelopeManifest -> dataEnvelopeFromBinary,
WriteManifest -> writeFromBinary,
WriteAckManifest -> (_ => WriteAck),
ReadManifest -> readFromBinary,
ReadResultManifest -> readResultFromBinary,
StatusManifest -> statusFromBinary,
GossipManifest -> gossipFromBinary,
DeltaPropagationManifest -> deltaPropagationFromBinary,
WriteNackManifest -> (_ => WriteNack),
DeltaNackManifest -> (_ => DeltaNack),
DurableDataEnvelopeManifest -> durableDataEnvelopeFromBinary)
override def manifest(obj: AnyRef): String = obj match {
case _: DataEnvelope => DataEnvelopeManifest
@ -272,9 +270,10 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
private def statusFromBinary(bytes: Array[Byte]): Status = {
val status = dm.Status.parseFrom(bytes)
Status(status.getEntriesList.asScala.iterator.map(e => e.getKey -> AkkaByteString(e.getDigest.toByteArray())).toMap,
status.getChunk,
status.getTotChunks)
Status(
status.getEntriesList.asScala.iterator.map(e => e.getKey -> AkkaByteString(e.getDigest.toByteArray())).toMap,
status.getChunk,
status.getTotChunks)
}
private def gossipToProto(gossip: Gossip): dm.Gossip = {
@ -288,8 +287,9 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
private def gossipFromBinary(bytes: Array[Byte]): Gossip = {
val gossip = dm.Gossip.parseFrom(decompress(bytes))
Gossip(gossip.getEntriesList.asScala.iterator.map(e => e.getKey -> dataEnvelopeFromProto(e.getEnvelope)).toMap,
sendBack = gossip.getSendBack)
Gossip(
gossip.getEntriesList.asScala.iterator.map(e => e.getKey -> dataEnvelopeFromProto(e.getEnvelope)).toMap,
sendBack = gossip.getSendBack)
}
private def deltaPropagationToProto(deltaPropagation: DeltaPropagation): dm.DeltaPropagation = {
@ -313,13 +313,14 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
private def deltaPropagationFromBinary(bytes: Array[Byte]): DeltaPropagation = {
val deltaPropagation = dm.DeltaPropagation.parseFrom(bytes)
val reply = deltaPropagation.hasReply && deltaPropagation.getReply
DeltaPropagation(uniqueAddressFromProto(deltaPropagation.getFromNode),
reply,
deltaPropagation.getEntriesList.asScala.iterator.map { e =>
val fromSeqNr = e.getFromSeqNr
val toSeqNr = if (e.hasToSeqNr) e.getToSeqNr else fromSeqNr
e.getKey -> Delta(dataEnvelopeFromProto(e.getEnvelope), fromSeqNr, toSeqNr)
}.toMap)
DeltaPropagation(
uniqueAddressFromProto(deltaPropagation.getFromNode),
reply,
deltaPropagation.getEntriesList.asScala.iterator.map { e =>
val fromSeqNr = e.getFromSeqNr
val toSeqNr = if (e.hasToSeqNr) e.getToSeqNr else fromSeqNr
e.getKey -> Delta(dataEnvelopeFromProto(e.getEnvelope), fromSeqNr, toSeqNr)
}.toMap)
}
private def getToProto(get: Get[_]): dm.Get = {

12
akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/DurableDataSpec.scala
View file

@ -98,8 +98,9 @@ abstract class DurableDataSpec(multiNodeConfig: DurableDataSpecConfig)
}
def newReplicator(sys: ActorSystem = system) =
sys.actorOf(Replicator.props(ReplicatorSettings(system).withGossipInterval(1.second)),
"replicator-" + testStepCounter)
sys.actorOf(
Replicator.props(ReplicatorSettings(system).withGossipInterval(1.second)),
"replicator-" + testStepCounter)
def join(from: RoleName, to: RoleName): Unit = {
runOn(from) {
@ -275,9 +276,10 @@ abstract class DurableDataSpec(multiNodeConfig: DurableDataSpecConfig)
Await.ready(sys1.terminate(), 10.seconds)
}
val sys2 = ActorSystem("AdditionalSys",
// use the same port
ConfigFactory.parseString(s"""
val sys2 = ActorSystem(
"AdditionalSys",
// use the same port
ConfigFactory.parseString(s"""
akka.remote.artery.canonical.port = ${address.port.get}
akka.remote.netty.tcp.port = ${address.port.get}
""").withFallback(system.settings.config))

16
akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/DurablePruningSpec.scala
View file

@ -49,11 +49,12 @@ class DurablePruningSpec extends MultiNodeSpec(DurablePruningSpec) with STMultiN
val maxPruningDissemination = 3.seconds
def startReplicator(sys: ActorSystem): ActorRef =
sys.actorOf(Replicator.props(
ReplicatorSettings(sys)
.withGossipInterval(1.second)
.withPruning(pruningInterval = 1.second, maxPruningDissemination)),
"replicator")
sys.actorOf(
Replicator.props(
ReplicatorSettings(sys)
.withGossipInterval(1.second)
.withPruning(pruningInterval = 1.second, maxPruningDissemination)),
"replicator")
val replicator = startReplicator(system)
val timeout = 5.seconds.dilated
@ -152,8 +153,9 @@ class DurablePruningSpec extends MultiNodeSpec(DurablePruningSpec) with STMultiN
runOn(first) {
val address = cluster2.selfAddress
val sys3 = ActorSystem(system.name,
ConfigFactory.parseString(s"""
val sys3 = ActorSystem(
system.name,
ConfigFactory.parseString(s"""
akka.remote.artery.canonical.port = ${address.port.get}
akka.remote.netty.tcp.port = ${address.port.get}
""").withFallback(system.settings.config))

14
akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/PerformanceSpec.scala
View file

@ -83,12 +83,14 @@ class PerformanceSpec extends MultiNodeSpec(PerformanceSpec) with STMultiNodeSpe
enterBarrier(from.name + "-joined")
}
def repeat(description: String,
keys: Iterable[ORSetKey[Int]],
n: Int,
expectedAfterReplication: Option[Set[Int]] = None,
oneByOne: Boolean = false)(block: (ORSetKey[Int], Int, ActorRef) => Unit,
afterEachKey: ORSetKey[Int] => Unit = _ => ()): Unit = {
def repeat(
description: String,
keys: Iterable[ORSetKey[Int]],
n: Int,
expectedAfterReplication: Option[Set[Int]] = None,
oneByOne: Boolean = false)(
block: (ORSetKey[Int], Int, ActorRef) => Unit,
afterEachKey: ORSetKey[Int] => Unit = _ => ()): Unit = {
keys.foreach { key =>
val startTime = System.nanoTime()

11
akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorPruningSpec.scala
View file

@ -41,11 +41,12 @@ class ReplicatorPruningSpec extends MultiNodeSpec(ReplicatorPruningSpec) with ST
val cluster = Cluster(system)
implicit val selfUniqueAddress = DistributedData(system).selfUniqueAddress
val maxPruningDissemination = 3.seconds
val replicator = system.actorOf(Replicator.props(
ReplicatorSettings(system)
.withGossipInterval(1.second)
.withPruning(pruningInterval = 1.second, maxPruningDissemination)),
"replicator")
val replicator = system.actorOf(
Replicator.props(
ReplicatorSettings(system)
.withGossipInterval(1.second)
.withPruning(pruningInterval = 1.second, maxPruningDissemination)),
"replicator")
val timeout = 3.seconds.dilated
val KeyA = GCounterKey("A")

40
akka-distributed-data/src/test/scala/akka/cluster/ddata/DeltaPropagationSelectorSpec.scala
View file

@ -53,9 +53,10 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
selector.hasDeltaEntries("A") should ===(true)
selector.hasDeltaEntries("B") should ===(true)
val expected =
DeltaPropagation(selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L), "B" -> Delta(DataEnvelope(deltaB), 1L, 1L)))
DeltaPropagation(
selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L), "B" -> Delta(DataEnvelope(deltaB), 1L, 1L)))
selector.collectPropagations() should ===(Map(nodes(0) -> expected))
selector.collectPropagations() should ===(Map.empty[Address, DeltaPropagation])
selector.cleanupDeltaEntries()
@ -68,9 +69,10 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
selector.update("A", deltaA)
selector.update("B", deltaB)
val expected =
DeltaPropagation(selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L), "B" -> Delta(DataEnvelope(deltaB), 1L, 1L)))
DeltaPropagation(
selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L), "B" -> Delta(DataEnvelope(deltaB), 1L, 1L)))
selector.collectPropagations() should ===(Map(nodes(0) -> expected, nodes(1) -> expected))
selector.cleanupDeltaEntries()
selector.hasDeltaEntries("A") should ===(true)
@ -87,17 +89,20 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
selector.update("A", deltaA)
selector.update("B", deltaB)
val expected1 =
DeltaPropagation(selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L), "B" -> Delta(DataEnvelope(deltaB), 1L, 1L)))
DeltaPropagation(
selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L), "B" -> Delta(DataEnvelope(deltaB), 1L, 1L)))
selector.collectPropagations() should ===(Map(nodes(0) -> expected1, nodes(1) -> expected1))
// new update before previous was propagated to all nodes
selector.update("C", deltaC)
val expected2 = DeltaPropagation(selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(deltaA), 1L, 1L),
"B" -> Delta(DataEnvelope(deltaB), 1L, 1L),
"C" -> Delta(DataEnvelope(deltaC), 1L, 1L)))
val expected2 = DeltaPropagation(
selfUniqueAddress,
false,
Map(
"A" -> Delta(DataEnvelope(deltaA), 1L, 1L),
"B" -> Delta(DataEnvelope(deltaB), 1L, 1L),
"C" -> Delta(DataEnvelope(deltaC), 1L, 1L)))
val expected3 = DeltaPropagation(selfUniqueAddress, false, Map("C" -> Delta(DataEnvelope(deltaC), 1L, 1L)))
selector.collectPropagations() should ===(Map(nodes(2) -> expected2, nodes(0) -> expected3))
selector.cleanupDeltaEntries()
@ -146,9 +151,10 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
selector.collectPropagations() should ===(Map(nodes(1) -> expected2))
selector.update("A", delta3)
val expected3 = DeltaPropagation(selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(delta1.merge(delta2).merge(delta3)), 1L, 3L)))
val expected3 = DeltaPropagation(
selfUniqueAddress,
false,
Map("A" -> Delta(DataEnvelope(delta1.merge(delta2).merge(delta3)), 1L, 3L)))
selector.collectPropagations() should ===(Map(nodes(2) -> expected3))
val expected4 =

5
akka-distributed-data/src/test/scala/akka/cluster/ddata/LocalConcurrencySpec.scala
View file

@ -47,8 +47,9 @@ class LocalConcurrencySpec(_system: ActorSystem)
def this() {
this(
ActorSystem("LocalConcurrencySpec",
ConfigFactory.parseString("""
ActorSystem(
"LocalConcurrencySpec",
ConfigFactory.parseString("""
akka.actor.provider = "cluster"
akka.remote.netty.tcp.port=0
akka.remote.artery.canonical.port = 0

7
akka-distributed-data/src/test/scala/akka/cluster/ddata/ORSetSpec.scala
View file

@ -511,9 +511,10 @@ class ORSetSpec extends WordSpec with Matchers {
"verify mergeDisjointKeys" in {
val keys: Set[Any] = Set("K3", "K4", "K5")
val elements: Map[Any, VersionVector] = Map("K3" -> VersionVector(nodeA, 4L),
"K4" -> VersionVector(TreeMap(nodeA -> 3L, nodeD -> 8L)),
"K5" -> VersionVector(nodeA, 2L))
val elements: Map[Any, VersionVector] = Map(
"K3" -> VersionVector(nodeA, 4L),
"K4" -> VersionVector(TreeMap(nodeA -> 3L, nodeD -> 8L)),
"K5" -> VersionVector(nodeA, 2L))
val vvector = VersionVector(TreeMap(nodeA -> 3L, nodeD -> 7L))
val acc: Map[Any, VersionVector] = Map("K1" -> VersionVector(nodeA, 3L))
val expectedDots = acc ++ Map("K3" -> VersionVector(nodeA, 4L), "K4" -> VersionVector(nodeD, 8L)) // "a" -> 3 removed, optimized to include only those unseen

102
akka-distributed-data/src/test/scala/akka/cluster/ddata/WriteAggregatorSpec.scala
View file

@ -24,34 +24,37 @@ object WriteAggregatorSpec {
val KeyA = GSetKey[String]("A")
val KeyB = ORSetKey[String]("B")
def writeAggregatorProps(data: GSet[String],
consistency: Replicator.WriteConsistency,
probes: Map[Address, ActorRef],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean): Props =
def writeAggregatorProps(
data: GSet[String],
consistency: Replicator.WriteConsistency,
probes: Map[Address, ActorRef],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean): Props =
Props(new TestWriteAggregator(KeyA, data, None, consistency, probes, nodes, unreachable, replyTo, durable))
def writeAggregatorPropsWithDelta(data: ORSet[String],
delta: Delta,
consistency: Replicator.WriteConsistency,
probes: Map[Address, ActorRef],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean): Props =
def writeAggregatorPropsWithDelta(
data: ORSet[String],
delta: Delta,
consistency: Replicator.WriteConsistency,
probes: Map[Address, ActorRef],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean): Props =
Props(new TestWriteAggregator(KeyB, data, Some(delta), consistency, probes, nodes, unreachable, replyTo, durable))
class TestWriteAggregator(key: Key.KeyR,
data: ReplicatedData,
delta: Option[Delta],
consistency: Replicator.WriteConsistency,
probes: Map[Address, ActorRef],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean)
class TestWriteAggregator(
key: Key.KeyR,
data: ReplicatedData,
delta: Option[Delta],
consistency: Replicator.WriteConsistency,
probes: Map[Address, ActorRef],
nodes: Set[Address],
unreachable: Set[Address],
replyTo: ActorRef,
durable: Boolean)
extends WriteAggregator(key, DataEnvelope(data), delta, consistency, None, nodes, unreachable, replyTo, durable) {
override def replica(address: Address): ActorSelection =
@ -215,14 +218,15 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
"send deltas first" in {
val probe = TestProbe()
val aggr = system.actorOf(
WriteAggregatorSpec.writeAggregatorPropsWithDelta(fullState2,
delta,
writeMajority,
probes(probe.ref),
nodes,
Set.empty,
testActor,
durable = false))
WriteAggregatorSpec.writeAggregatorPropsWithDelta(
fullState2,
delta,
writeMajority,
probes(probe.ref),
nodes,
Set.empty,
testActor,
durable = false))
probe.expectMsgType[DeltaPropagation]
probe.lastSender ! WriteAck
@ -237,14 +241,15 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
val testProbes = probes()
val testProbeRefs = testProbes.map { case (a, tm) => a -> tm.writeAckAdapter }
val aggr = system.actorOf(
WriteAggregatorSpec.writeAggregatorPropsWithDelta(fullState2,
delta,
writeAll,
testProbeRefs,
nodes,
Set.empty,
testActor,
durable = false))
WriteAggregatorSpec.writeAggregatorPropsWithDelta(
fullState2,
delta,
writeAll,
testProbeRefs,
nodes,
Set.empty,
testActor,
durable = false))
testProbes(nodeA).expectMsgType[DeltaPropagation]
// no reply
@ -271,14 +276,15 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
"timeout when less than required acks" in {
val probe = TestProbe()
val aggr = system.actorOf(
WriteAggregatorSpec.writeAggregatorPropsWithDelta(fullState2,
delta,
writeAll,
probes(probe.ref),
nodes,
Set.empty,
testActor,
durable = false))
WriteAggregatorSpec.writeAggregatorPropsWithDelta(
fullState2,
delta,
writeAll,
probes(probe.ref),
nodes,
Set.empty,
testActor,
durable = false))
probe.expectMsgType[DeltaPropagation]
// no reply

30
akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatedDataSerializerSpec.scala
View file

@ -28,8 +28,9 @@ import akka.testkit.TestActors
class ReplicatedDataSerializerSpec
extends TestKit(
ActorSystem("ReplicatedDataSerializerSpec",
ConfigFactory.parseString("""
ActorSystem(
"ReplicatedDataSerializerSpec",
ConfigFactory.parseString("""
akka.loglevel = DEBUG
akka.actor.provider=cluster
akka.remote.netty.tcp.port=0
@ -202,10 +203,12 @@ class ReplicatedDataSerializerSpec
checkSerialization(GCounter().increment(address1, 3))
checkSerialization(GCounter().increment(address1, 2).increment(address2, 5))
checkSameContent(GCounter().increment(address1, 2).increment(address2, 5),
GCounter().increment(address2, 5).increment(address1, 1).increment(address1, 1))
checkSameContent(GCounter().increment(address1, 2).increment(address3, 5),
GCounter().increment(address3, 5).increment(address1, 2))
checkSameContent(
GCounter().increment(address1, 2).increment(address2, 5),
GCounter().increment(address2, 5).increment(address1, 1).increment(address1, 1))
checkSameContent(
GCounter().increment(address1, 2).increment(address3, 5),
GCounter().increment(address3, 5).increment(address1, 2))
}
"serialize PNCounter" in {
@ -215,12 +218,15 @@ class ReplicatedDataSerializerSpec
checkSerialization(PNCounter().increment(address1, 2).increment(address2, 5))
checkSerialization(PNCounter().increment(address1, 2).increment(address2, 5).decrement(address1, 1))
checkSameContent(PNCounter().increment(address1, 2).increment(address2, 5),
PNCounter().increment(address2, 5).increment(address1, 1).increment(address1, 1))
checkSameContent(PNCounter().increment(address1, 2).increment(address3, 5),
PNCounter().increment(address3, 5).increment(address1, 2))
checkSameContent(PNCounter().increment(address1, 2).decrement(address1, 1).increment(address3, 5),
PNCounter().increment(address3, 5).increment(address1, 2).decrement(address1, 1))
checkSameContent(
PNCounter().increment(address1, 2).increment(address2, 5),
PNCounter().increment(address2, 5).increment(address1, 1).increment(address1, 1))
checkSameContent(
PNCounter().increment(address1, 2).increment(address3, 5),
PNCounter().increment(address3, 5).increment(address1, 2))
checkSameContent(
PNCounter().increment(address1, 2).decrement(address1, 1).increment(address3, 5),
PNCounter().increment(address3, 5).increment(address1, 2).decrement(address1, 1))
}
"serialize ORMap" in {

32
akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializerSpec.scala
View file

@ -31,8 +31,9 @@ import akka.cluster.ddata.ORMultiMap
class ReplicatorMessageSerializerSpec
extends TestKit(
ActorSystem("ReplicatorMessageSerializerSpec",
ConfigFactory.parseString("""
ActorSystem(
"ReplicatorMessageSerializerSpec",
ConfigFactory.parseString("""
akka.actor.provider=cluster
akka.remote.netty.tcp.port=0
akka.remote.artery.canonical.port = 0
@ -87,9 +88,11 @@ class ReplicatorMessageSerializerSpec
checkSerialization(Changed(keyA)(data1))
checkSerialization(DataEnvelope(data1))
checkSerialization(
DataEnvelope(data1,
pruning = Map(address1 -> PruningPerformed(System.currentTimeMillis()),
address3 -> PruningInitialized(address2, Set(address1.address)))))
DataEnvelope(
data1,
pruning = Map(
address1 -> PruningPerformed(System.currentTimeMillis()),
address3 -> PruningInitialized(address2, Set(address1.address)))))
checkSerialization(Write("A", DataEnvelope(data1)))
checkSerialization(WriteAck)
checkSerialization(WriteNack)
@ -102,16 +105,19 @@ class ReplicatorMessageSerializerSpec
checkSerialization(
Gossip(Map("A" -> DataEnvelope(data1), "B" -> DataEnvelope(GSet() + "b" + "c")), sendBack = true))
checkSerialization(
DeltaPropagation(address1,
reply = true,
Map("A" -> Delta(DataEnvelope(delta1), 1L, 1L),
"B" -> Delta(DataEnvelope(delta2), 3L, 5L),
"C" -> Delta(DataEnvelope(delta3), 1L, 1L),
"DC" -> Delta(DataEnvelope(delta4), 1L, 1L))))
DeltaPropagation(
address1,
reply = true,
Map(
"A" -> Delta(DataEnvelope(delta1), 1L, 1L),
"B" -> Delta(DataEnvelope(delta2), 3L, 5L),
"C" -> Delta(DataEnvelope(delta3), 1L, 1L),
"DC" -> Delta(DataEnvelope(delta4), 1L, 1L))))
checkSerialization(new DurableDataEnvelope(data1))
val pruning = Map(address1 -> PruningPerformed(System.currentTimeMillis()),
address3 -> PruningInitialized(address2, Set(address1.address)))
val pruning = Map(
address1 -> PruningPerformed(System.currentTimeMillis()),
address3 -> PruningInitialized(address2, Set(address1.address)))
val deserializedDurableDataEnvelope =
checkSerialization(
new DurableDataEnvelope(DataEnvelope(data1, pruning, deltaVersions = VersionVector(address1, 13L))))