import akka.actor.ActorRef;
import akka.actor.Cancellable;
+import com.google.common.base.Preconditions;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
*/
private Cancellable electionCancel = null;
- /**
- *
- */
- protected String leaderId = null;
-
- private short leaderPayloadVersion = -1;
-
private long replicatedToAllIndex = -1;
private final String logName;
private final RaftState state;
- protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) {
- this.context = context;
- this.state = state;
+ AbstractRaftActorBehavior(final RaftActorContext context, final RaftState state) {
+ this.context = Preconditions.checkNotNull(context);
+ this.state = Preconditions.checkNotNull(state);
this.LOG = context.getLogger();
logName = String.format("%s (%s)", context.getId(), state);
}
+ public static RaftActorBehavior createBehavior(final RaftActorContext context, final RaftState state) {
+ switch (state) {
+ case Candidate:
+ return new Candidate(context);
+ case Follower:
+ return new Follower(context);
+ case IsolatedLeader:
+ return new IsolatedLeader(context);
+ case Leader:
+ return new Leader(context);
+ default:
+ throw new IllegalArgumentException("Unhandled state " + state);
+ }
+ }
+
@Override
- public RaftState state() {
+ public final RaftState state() {
return state;
}
- public String logName() {
+ protected final String logName() {
return logName;
}
return context.getReplicatedLog().lastIndex();
}
- /**
- * @param logIndex
- * @return the client request tracker for the specified logIndex
- */
- protected ClientRequestTracker findClientRequestTracker(long logIndex) {
- return null;
- }
-
/**
* @param logIndex
* @return the client request tracker for the specified logIndex
return null;
}
-
/**
*
* @return log index from the previous to last entry in the log
protected void applyLogToStateMachine(final long index) {
long newLastApplied = context.getLastApplied();
// Now maybe we apply to the state machine
- for (long i = context.getLastApplied() + 1;
- i < index + 1; i++) {
- ActorRef clientActor = null;
- String identifier = null;
- ClientRequestTracker tracker = removeClientRequestTracker(i);
-
- if (tracker != null) {
- clientActor = tracker.getClientActor();
- identifier = tracker.getIdentifier();
- }
- ReplicatedLogEntry replicatedLogEntry =
- context.getReplicatedLog().get(i);
+ for (long i = context.getLastApplied() + 1; i < index + 1; i++) {
+ ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(i);
if (replicatedLogEntry != null) {
// Send a local message to the local RaftActor (it's derived class to be
// specific to apply the log to it's index)
- actor().tell(new ApplyState(clientActor, identifier,
- replicatedLogEntry), actor());
+
+ final ApplyState msg;
+ final ClientRequestTracker tracker = removeClientRequestTracker(i);
+ if (tracker != null) {
+ msg = new ApplyState(tracker.getClientActor(), tracker.getIdentifier(), replicatedLogEntry);
+ } else {
+ msg = new ApplyState(null, null, replicatedLogEntry);
+ }
+
+ actor().tell(msg, actor());
newLastApplied = i;
} else {
//if one index is not present in the log, no point in looping
// around as the rest wont be present either
- LOG.warn(
- "{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}",
+ LOG.warn("{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}",
logName(), i, i, index);
break;
}
return requestVote(sender, (RequestVote) message);
} else if (message instanceof RequestVoteReply) {
return handleRequestVoteReply(sender, (RequestVoteReply) message);
+ } else {
+ return null;
}
- return this;
- }
-
- @Override public String getLeaderId() {
- return leaderId;
- }
-
- @Override
- public short getLeaderPayloadVersion() {
- return leaderPayloadVersion;
- }
-
- public void setLeaderPayloadVersion(short leaderPayloadVersion) {
- this.leaderPayloadVersion = leaderPayloadVersion;
}
@Override
protected RaftActorBehavior internalSwitchBehavior(RaftState newState) {
if(context.getRaftPolicy().automaticElectionsEnabled()){
- return internalSwitchBehavior(newState.createBehavior(context));
+ return internalSwitchBehavior(createBehavior(context, newState));
}
return this;
}
* @param snapshotCapturedIndex
*/
protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) {
- long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex, this);
+ long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex);
if(actualIndex != -1){
setReplicatedToAllIndex(actualIndex);