import akka.persistence.SnapshotOffer;
import akka.persistence.SnapshotSelectionCriteria;
import akka.persistence.UntypedPersistentActor;
+import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
+import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
+import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
import org.opendaylight.controller.cluster.raft.behaviors.Candidate;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
import org.opendaylight.controller.cluster.raft.behaviors.Leader;
import org.opendaylight.controller.cluster.raft.client.messages.AddRaftPeer;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
-import org.opendaylight.controller.cluster.raft.internal.messages.ApplySnapshot;
import org.opendaylight.controller.cluster.raft.client.messages.RemoveRaftPeer;
-import org.opendaylight.controller.cluster.raft.internal.messages.ApplyState;
-import org.opendaylight.controller.cluster.raft.internal.messages.Replicate;
+import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import java.io.Serializable;
import java.util.ArrayList;
applyState(applyState.getClientActor(), applyState.getIdentifier(),
applyState.getReplicatedLogEntry().getData());
+
} else if(message instanceof ApplySnapshot ) {
applySnapshot(((ApplySnapshot) message).getSnapshot());
+
} else if (message instanceof FindLeader) {
getSender().tell(
new FindLeaderReply(
context.getPeerAddress(currentBehavior.getLeaderId())),
getSelf()
);
+
} else if (message instanceof SaveSnapshotSuccess) {
SaveSnapshotSuccess success = (SaveSnapshotSuccess) message;
} else if (message instanceof SaveSnapshotFailure) {
// TODO: Handle failure in saving the snapshot
- // Maybe do retries on failure
+
+ } else if (message instanceof FindLeader){
+
+ getSender().tell(new FindLeaderReply(
+ context.getPeerAddress(currentBehavior.getLeaderId())),
+ getSelf());
} else if (message instanceof AddRaftPeer){
* @param data
*/
protected void persistData(ActorRef clientActor, String identifier,
- Object data) {
+ Payload data) {
ReplicatedLogEntry replicatedLogEntry = new ReplicatedLogImplEntry(
context.getReplicatedLog().lastIndex() + 1,
return currentBehavior.state();
}
+ /**
+ * setPeerAddress sets the address of a known peer at a later time.
+ * <p>
+ * This is to account for situations where a we know that a peer
+ * exists but we do not know an address up-front. This may also be used in
+ * situations where a known peer starts off in a different location and we
+ * need to change it's address
+ * <p>
+ * Note that if the peerId does not match the list of peers passed to
+ * this actor during construction an IllegalStateException will be thrown.
+ *
+ * @param peerId
+ * @param peerAddress
+ */
+ protected void setPeerAddress(String peerId, String peerAddress){
+ context.setPeerAddress(peerId, peerAddress);
+ }
+
/**
final String identifier,
final ReplicatedLogEntry replicatedLogEntry) {
context.getLogger().debug(
- "Append log entry and persist " + replicatedLogEntry);
+ "Append log entry and persist {} ", replicatedLogEntry);
// FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
journal.add(replicatedLogEntry);
}
- private static class ReplicatedLogImplEntry implements ReplicatedLogEntry,
- Serializable {
-
- private final long index;
- private final long term;
- private final Object payload;
- public ReplicatedLogImplEntry(long index, long term, Object payload) {
-
- this.index = index;
- this.term = term;
- this.payload = payload;
- }
-
- @Override public Object getData() {
- return payload;
- }
-
- @Override public long getTerm() {
- return term;
- }
-
- @Override public long getIndex() {
- return index;
- }
-
- @Override public String toString() {
- return "Entry{" +
- "index=" + index +
- ", term=" + term +
- '}';
- }
- }
private static class DeleteEntries implements Serializable {
private final int fromIndex;