import org.opendaylight.controller.cluster.raft.internal.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
import org.opendaylight.controller.cluster.raft.messages.RequestVote;
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
}
@Override protected RaftState handleAppendEntries(ActorRef sender,
- AppendEntries appendEntries, RaftState suggestedState) {
+ AppendEntries appendEntries) {
- context.getLogger().error("An unexpected AppendEntries received in state " + state());
-
- return suggestedState;
+ return state();
}
@Override protected RaftState handleAppendEntriesReply(ActorRef sender,
- AppendEntriesReply appendEntriesReply, RaftState suggestedState) {
-
- // Some peer thinks I was a leader and sent me a reply
+ AppendEntriesReply appendEntriesReply) {
- return suggestedState;
+ return state();
}
@Override protected RaftState handleRequestVoteReply(ActorRef sender,
- RequestVoteReply requestVoteReply, RaftState suggestedState) {
- if (suggestedState == RaftState.Follower) {
- // If base class thinks I should be follower then I am
- return suggestedState;
- }
+ RequestVoteReply requestVoteReply) {
if (requestVoteReply.isVoteGranted()) {
voteCount++;
@Override
public RaftState handleMessage(ActorRef sender, Object message) {
+
+ if (message instanceof RaftRPC) {
+ RaftRPC rpc = (RaftRPC) message;
+ // If RPC request or response contains term T > currentTerm:
+ // set currentTerm = T, convert to follower (ยง5.1)
+ // This applies to all RPC messages and responses
+ if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
+ context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
+ return RaftState.Follower;
+ }
+ }
+
if (message instanceof ElectionTimeout) {
if (votesRequired == 0) {
// If there are no peers then we should be a Leader
// Increment the election term and vote for self
long currentTerm = context.getTermInformation().getCurrentTerm();
- context.getTermInformation().update(currentTerm + 1, context.getId());
+ context.getTermInformation().updateAndPersist(currentTerm + 1, context.getId());
context.getLogger().debug("Starting new term " + (currentTerm+1));
// Request for a vote
+ // TODO: Retry request for vote if replies do not arrive in a reasonable
+ // amount of time TBD
for (ActorSelection peerActor : peerToActor.values()) {
peerActor.tell(new RequestVote(
context.getTermInformation().getCurrentTerm(),