diff --git a/akka-cluster/src/main/scala/akka/cluster/Cluster.scala b/akka-cluster/src/main/scala/akka/cluster/Cluster.scala index 415a50a98a..48c77ce7db 100644 --- a/akka-cluster/src/main/scala/akka/cluster/Cluster.scala +++ b/akka-cluster/src/main/scala/akka/cluster/Cluster.scala @@ -435,10 +435,10 @@ class DefaultClusterNode private[akka] ( /** * Registers a cluster change listener. */ - def register(listener: ChangeListener): ClusterNode = if (isConnected.isOff) { + def register(listener: ChangeListener): ClusterNode = { changeListeners.add(listener) this - } else throw new IllegalStateException("Can not register 'ChangeListener' after the cluster node has been started") + } private[cluster] def publish(change: ChangeNotification) { changeListeners.iterator.foreach(_.notify(change, this)) diff --git a/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala b/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala index aa6016aa15..4bd722f05f 100644 --- a/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala +++ b/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala @@ -17,8 +17,6 @@ import java.util.concurrent.{ CyclicBarrier, TimeUnit } import scala.collection.JavaConversions._ -// FIXME: Test sending all funs - class MyJavaSerializableActor extends Actor with Serializable { var count = 0 diff --git a/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode1.conf b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode1.conf new file mode 100644 index 0000000000..480c30c09d --- /dev/null +++ b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode1.conf @@ -0,0 +1 @@ +akka.event-handler-level = "DEBUG" diff --git a/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode1.opts b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode1.opts new file mode 100644 index 0000000000..a88c260d8c --- /dev/null +++ b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode1.opts @@ -0,0 +1 @@ +-Dakka.cluster.nodename=node1 -Dakka.cluster.port=9991 diff --git a/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode2.conf b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode2.conf new file mode 100644 index 0000000000..480c30c09d --- /dev/null +++ b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode2.conf @@ -0,0 +1 @@ +akka.event-handler-level = "DEBUG" diff --git a/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode2.opts b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode2.opts new file mode 100644 index 0000000000..f1e01f253d --- /dev/null +++ b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmNode2.opts @@ -0,0 +1 @@ +-Dakka.cluster.nodename=node2 -Dakka.cluster.port=9992 diff --git a/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmSpec.scala b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmSpec.scala new file mode 100644 index 0000000000..b035a33ec7 --- /dev/null +++ b/akka-cluster/src/test/scala/akka/cluster/changelisteners/nodeconnected/NodeConnectedChangeListenerMultiJvmSpec.scala @@ -0,0 +1,81 @@ +/** + * Copyright (C) 2009-2011 Scalable Solutions AB + */ + +package akka.cluster.changelisteners.nodeconnected + +import org.scalatest.WordSpec +import org.scalatest.matchers.MustMatchers +import org.scalatest.BeforeAndAfterAll + +import akka.cluster._ +import ChangeListener._ + +import java.util.concurrent._ + +object NodeConnectedChangeListenerMultiJvmSpec { + var NrOfNodes = 2 +} + +class NodeConnectedChangeListenerMultiJvmNode1 extends WordSpec with MustMatchers with BeforeAndAfterAll { + import NodeConnectedChangeListenerMultiJvmSpec._ + + "A NodeConnected change listener" must { + + "be invoked when a new node joins the cluster" in { + System.getProperty("akka.cluster.nodename", "") must be("node1") + System.getProperty("akka.cluster.port", "") must be("9991") + + val latch = new CountDownLatch(1) + Cluster.node.register(new ChangeListener { + override def nodeConnected(node: String, client: ClusterNode) { + latch.countDown + } + }) + + Cluster.node.shutdown() + + Cluster.barrier("start-node1", NrOfNodes) { + Cluster.node.start() + } + + Cluster.barrier("start-node2", NrOfNodes) { + latch.await(5, TimeUnit.SECONDS) must be === true + } + + Cluster.barrier("shutdown", NrOfNodes) { + // Cluster.node.shutdown() + } + } + } + + override def beforeAll() = { + Cluster.startLocalCluster() + } + + override def afterAll() = { + Cluster.shutdownLocalCluster() + } +} + +class NodeConnectedChangeListenerMultiJvmNode2 extends WordSpec with MustMatchers { + import NodeConnectedChangeListenerMultiJvmSpec._ + + "A NodeConnected change listener" must { + + "be invoked when a new node joins the cluster" in { + System.getProperty("akka.cluster.nodename", "") must be("node2") + System.getProperty("akka.cluster.port", "") must be("9992") + + Cluster.barrier("start-node1", NrOfNodes) {} + + Cluster.barrier("start-node2", NrOfNodes) { + Cluster.node.start() + } + + Cluster.barrier("shutdown", NrOfNodes) { + // Cluster.node.shutdown() + } + } + } +}