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

Merge pull request #1480 from akka/wip-3388-HeartbeatReq-patriknw

Browse source 
Start heartbeatSender after Welcome, see #3388
This commit is contained in:
Patrik Nordwall 2013-05-27 00:17:43 -07:00
commit 28d1b1f187
1 changed files with 10 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

15
akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
View file

@ -242,9 +242,6 @@ private[cluster] final class ClusterCoreDaemon(publisher: ActorRef) extends Acto
private def clusterCore(address: Address): ActorSelection = private def clusterCore(address: Address): ActorSelection =
context.actorSelection(RootActorPath(address) / "system" / "cluster" / "core" / "daemon") context.actorSelection(RootActorPath(address) / "system" / "cluster" / "core" / "daemon")
context.actorOf(Props[ClusterHeartbeatSender].
withDispatcher(UseDispatcher), name = "heartbeatSender")
import context.dispatcher import context.dispatcher
// start periodic gossip to random nodes in cluster // start periodic gossip to random nodes in cluster
@ -304,6 +301,14 @@ private[cluster] final class ClusterCoreDaemon(publisher: ActorRef) extends Acto
} }
} }
def becomeInitialized(): Unit = {
// start heartbeatSender here, and not in constructor to make sure that
// heartbeating doesn't start before Welcome is received
context.actorOf(Props[ClusterHeartbeatSender].
withDispatcher(UseDispatcher), name = "heartbeatSender")
context.become(initialized)
}
def initialized: Actor.Receive = { def initialized: Actor.Receive = {
case msg: GossipEnvelope ⇒ receiveGossip(msg) case msg: GossipEnvelope ⇒ receiveGossip(msg)
case msg: GossipStatus ⇒ receiveGossipStatus(msg) case msg: GossipStatus ⇒ receiveGossipStatus(msg)
@ -384,7 +389,7 @@ private[cluster] final class ClusterCoreDaemon(publisher: ActorRef) extends Acto
} }
if (address == selfAddress) { if (address == selfAddress) {
context.become(initialized) becomeInitialized()
joining(selfUniqueAddress, cluster.selfRoles) joining(selfUniqueAddress, cluster.selfRoles)
} else { } else {
val joinDeadline = RetryUnsuccessfulJoinAfter match { val joinDeadline = RetryUnsuccessfulJoinAfter match {
@ -457,7 +462,7 @@ private[cluster] final class ClusterCoreDaemon(publisher: ActorRef) extends Acto
publish(latestGossip) publish(latestGossip)
if (from != selfUniqueAddress) if (from != selfUniqueAddress)
gossipTo(from, sender) gossipTo(from, sender)
context.become(initialized) becomeInitialized()
} }
} }