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

+con #3677 Limit delta in DistributedPubSubMediator

Browse source
This commit is contained in:
Patrik Nordwall 2013-10-21 12:00:38 +02:00
parent ff83edea0b
commit 11972b4497
4 changed files with 79 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

46
akka-contrib/src/multi-jvm/scala/akka/contrib/pattern/DistributedPubSubMediatorSpec.scala
View file

@ -18,6 +18,8 @@ import akka.remote.testkit.MultiNodeSpec
import akka.remote.testkit.STMultiNodeSpec
import akka.testkit._
import akka.actor.ActorLogging
import akka.contrib.pattern.DistributedPubSubMediator.Internal.Status
import akka.contrib.pattern.DistributedPubSubMediator.Internal.Delta
object DistributedPubSubMediatorSpec extends MultiNodeConfig {
val first = role("first")
@ -29,6 +31,8 @@ object DistributedPubSubMediatorSpec extends MultiNodeConfig {
akka.actor.provider = "akka.cluster.ClusterActorRefProvider"
akka.remote.log-remote-lifecycle-events = off
akka.cluster.auto-down-unreachable-after = 0s
akka.contrib.cluster.pub-sub.max-delta-elements = 500
#akka.remote.log-frame-size-exceeding = 1024b
"""))
object TestChatUser {
@ -341,5 +345,47 @@ class DistributedPubSubMediatorSpec extends MultiNodeSpec(DistributedPubSubMedia
enterBarrier("after-11")
}
"transfer delta correctly" in {
val firstAddress = node(first).address
val secondAddress = node(second).address
val thirdAddress = node(third).address
runOn(first) {
mediator ! Status(versions = Map.empty)
val deltaBuckets = expectMsgType[Delta].buckets
deltaBuckets.size must be (3)
deltaBuckets.find(_.owner == firstAddress).get.content.size must be(7)
deltaBuckets.find(_.owner == secondAddress).get.content.size must be(6)
deltaBuckets.find(_.owner == thirdAddress).get.content.size must be(2)
}
enterBarrier("verified-initial-delta")
// this test is configured with max-delta-elements = 500
val many = 1010
runOn(first) {
for (i <- 0 until many)
mediator ! Put(createChatUser("u" + (1000 + i)))
mediator ! Status(versions = Map.empty)
val deltaBuckets1 = expectMsgType[Delta].buckets
deltaBuckets1.map(_.content.size).sum must be (500)
mediator ! Status(versions = deltaBuckets1.map(b => b.owner -> b.version).toMap)
val deltaBuckets2 = expectMsgType[Delta].buckets
deltaBuckets1.map(_.content.size).sum must be (500)
mediator ! Status(versions = deltaBuckets2.map(b => b.owner -> b.version).toMap)
val deltaBuckets3 = expectMsgType[Delta].buckets
deltaBuckets3.map(_.content.size).sum must be (7 + 6 + 2 + many - 500 - 500)
}
enterBarrier("verified-delta-with-many")
awaitCount(13 + many)
enterBarrier("after-12")
}
}
}