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

ZK related fixed. No review since there's nothing to review.

Browse source
This commit is contained in:
Eugene Vigdorchik 2011-12-30 17:46:04 +04:00
parent 17125af571
commit 7b9e9d2fe0
2 changed files with 38 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-remote/src/multi-jvm/scala/akka/remote/MultiJvmSync.scala
View file

@ -43,7 +43,7 @@ object MultiJvmSync {
val barrier = if (AkkaRemoteSpec.testNodes eq null) val barrier = if (AkkaRemoteSpec.testNodes eq null)
new FileBasedBarrier(name, count, testName, nodeName, timeout) new FileBasedBarrier(name, count, testName, nodeName, timeout)
else else
new ZkClient.ZkBarrier(name + "_" + nodeName, count, testName) new ZkClient.ZkBarrier(name + "_" + nodeName, count, "/" + testName)
barrier.await() barrier.await()
} }
} }

45
akka-remote/src/multi-jvm/scala/akka/remote/ZKClient.scala
View file

@ -6,12 +6,14 @@ package akka.remote
import com.typesafe.config.Config import com.typesafe.config.Config
import org.apache.zookeeper._ import org.apache.zookeeper._
import ZooDefs.Ids import ZooDefs.Ids
import collection.JavaConversions._
import java.net.InetAddress
object ZkClient extends Watcher { object ZkClient extends Watcher {
// Don't forget to close! // Don't forget to close!
lazy val zk: ZooKeeper = { lazy val zk: ZooKeeper = {
val remoteNodes = AkkaRemoteSpec.testNodes split ',' val remoteNodes = AkkaRemoteSpec.testNodes split ','
// ZkServers are configured to listen on a specific port. // ZkServers are configured to listen on a specific port.
val connectString = remoteNodes map (_+":2181") mkString "," val connectString = remoteNodes map (_+":2181") mkString ","
new ZooKeeper(connectString, 3000, this) new ZooKeeper(connectString, 3000, this)
@ -22,29 +24,56 @@ object ZkClient extends Watcher {
} }
class ZkBarrier(name: String, count: Int, root: String) extends Barrier { class ZkBarrier(name: String, count: Int, root: String) extends Barrier {
if (zk.exists(root, false) eq null) { @annotation.tailrec
zk.create(root, Array[Byte](), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL) private def waitForServer() {
// SI-1672
val r = try {
zk.exists("/", false); true
} catch {
case _: KeeperException.ConnectionLossException =>
println("Server is not ready, sleeping...")
Thread.sleep(10000)
false
}
if (!r) waitForServer()
}
waitForServer()
try {
zk.create(root, Array[Byte](), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
} catch {
case _: KeeperException.NodeExistsException =>
} }
def enter() { def enter() {
zk.create(root + "/" + name, Array[Byte](), Ids.OPEN_ACL_UNSAFE, zk.create(root + "/" + name, Array[Byte](), Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL) CreateMode.EPHEMERAL)
while (true) { while (true) {
ZkClient.this.synchronized { ZkClient.this.synchronized {
if (zk.getChildren(root, true).size < count) { val children = zk.getChildren(root, true)
println("Enter, children: " + children.mkString(","))
if (children.size < count) {
println("waiting")
ZkClient.this.wait() ZkClient.this.wait()
} } else
return
} }
} }
} }
def leave() { final def leave() {
zk.delete(root + "/" + name, -1) zk.delete(root + "/" + name, -1)
while (true) { while (true) {
ZkClient.this.synchronized { ZkClient.this.synchronized {
if (!zk.getChildren(root, true).isEmpty) { val children = zk.getChildren(root, true)
println("Leave, children: " + children.mkString(","))
if (!children.isEmpty) {
println("waiting")
ZkClient.this.wait() ZkClient.this.wait()
} } else
return
} }
} }
} }