import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.actor.Props;
-import com.google.common.base.Preconditions;
-import com.google.protobuf.GeneratedMessage;
+import akka.japi.Procedure;
+import com.google.common.base.Throwables;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
+import org.opendaylight.controller.cluster.NonPersistentDataProvider;
+import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
+import org.opendaylight.controller.cluster.raft.policy.RaftPolicy;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
-import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
-import org.opendaylight.controller.protobuff.messages.cluster.raft.test.MockPayloadMessages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class MockRaftActorContext implements RaftActorContext {
+public class MockRaftActorContext extends RaftActorContextImpl {
+ private static final Logger LOG = LoggerFactory.getLogger(MockRaftActorContext.class);
- private String id;
private ActorSystem system;
- private ActorRef actor;
- private long index = 0;
- private long lastApplied = 0;
- private final ElectionTerm electionTerm;
- private ReplicatedLog replicatedLog;
- private Map<String, String> peerAddresses = new HashMap<>();
- private ConfigParams configParams;
- private boolean snapshotCaptureInitiated;
-
- public MockRaftActorContext(){
- electionTerm = null;
+ private RaftPolicy raftPolicy;
- initReplicatedLog();
- }
-
- public MockRaftActorContext(String id, ActorSystem system, ActorRef actor){
- this.id = id;
- this.system = system;
- this.actor = actor;
-
- final String id1 = id;
- electionTerm = new ElectionTerm() {
- /**
- * Identifier of the actor whose election term information this is
- */
- private final String id = id1;
+ private static ElectionTerm newElectionTerm() {
+ return new ElectionTerm() {
private long currentTerm = 1;
private String votedFor = "";
}
@Override
- public void update(long currentTerm, String votedFor){
- this.currentTerm = currentTerm;
- this.votedFor = votedFor;
+ public void update(long newTerm, String newVotedFor) {
+ this.currentTerm = newTerm;
+ this.votedFor = newVotedFor;
// TODO : Write to some persistent state
}
- @Override public void updateAndPersist(long currentTerm,
- String votedFor) {
- update(currentTerm, votedFor);
+ @Override public void updateAndPersist(long newTerm, String newVotedFor) {
+ update(newTerm, newVotedFor);
}
};
+ }
- configParams = new DefaultConfigParamsImpl();
+ public MockRaftActorContext() {
+ super(null, null, "test", newElectionTerm(), -1, -1, new HashMap<>(),
+ new DefaultConfigParamsImpl(), new NonPersistentDataProvider(), LOG);
+ setReplicatedLog(new MockReplicatedLogBuilder().build());
+ }
+
+ public MockRaftActorContext(String id, ActorSystem system, ActorRef actor) {
+ super(actor, null, id, newElectionTerm(), -1, -1, new HashMap<>(),
+ new DefaultConfigParamsImpl(), new NonPersistentDataProvider(), LOG);
+
+ this.system = system;
initReplicatedLog();
}
- public void initReplicatedLog(){
- this.replicatedLog = new SimpleReplicatedLog();
+ public void initReplicatedLog() {
+ SimpleReplicatedLog replicatedLog = new SimpleReplicatedLog();
long term = getTermInformation().getCurrentTerm();
- this.replicatedLog.append(new MockReplicatedLogEntry(term, 0, new MockPayload("1")));
- this.replicatedLog.append(new MockReplicatedLogEntry(term, 1, new MockPayload("2")));
+ replicatedLog.append(new MockReplicatedLogEntry(term, 0, new MockPayload("1")));
+ replicatedLog.append(new MockReplicatedLogEntry(term, 1, new MockPayload("2")));
+ setReplicatedLog(replicatedLog);
+ setCommitIndex(replicatedLog.lastIndex());
+ setLastApplied(replicatedLog.lastIndex());
}
@Override public ActorRef actorOf(Props props) {
return system.actorSelection(path);
}
- @Override public String getId() {
- return id;
- }
-
- @Override public ActorRef getActor() {
- return actor;
- }
-
- @Override public ElectionTerm getTermInformation() {
- return electionTerm;
- }
-
- public void setIndex(long index){
- this.index = index;
- }
-
- @Override public long getCommitIndex() {
- return index;
- }
-
- @Override public void setCommitIndex(long commitIndex) {
- this.index = commitIndex;
- }
-
- @Override public void setLastApplied(long lastApplied){
- this.lastApplied = lastApplied;
- }
-
- @Override public long getLastApplied() {
- return lastApplied;
- }
-
- @Override
- // FIXME : A lot of tests try to manipulate the replicated log by setting it using this method
- // This is OK to do if the underlyingActor is not RafActor or a derived class. If not then you should not
- // used this way to manipulate the log because the RaftActor actually has a field replicatedLog
- // which it creates internally and sets on the RaftActorContext
- // The only right way to manipulate the replicated log therefore is to get it from either the RaftActor
- // or the RaftActorContext and modify the entries in there instead of trying to replace it by using this setter
- // Simple assertion that will fail if you do so
- // ReplicatedLog log = new ReplicatedLogImpl();
- // raftActor.underlyingActor().getRaftActorContext().setReplicatedLog(log);
- // assertEquals(log, raftActor.underlyingActor().getReplicatedLog())
- public void setReplicatedLog(ReplicatedLog replicatedLog) {
- this.replicatedLog = replicatedLog;
- }
-
- @Override public ReplicatedLog getReplicatedLog() {
- return replicatedLog;
- }
-
@Override public ActorSystem getActorSystem() {
return this.system;
}
- @Override public Logger getLogger() {
- return LoggerFactory.getLogger(getClass());
- }
-
- @Override public Map<String, String> getPeerAddresses() {
- return peerAddresses;
- }
-
- @Override public String getPeerAddress(String peerId) {
- return peerAddresses.get(peerId);
- }
-
- @Override public void addToPeers(String name, String address) {
- peerAddresses.put(name, address);
- }
-
- @Override public void removePeer(String name) {
- peerAddresses.remove(name);
- }
-
@Override public ActorSelection getPeerActorSelection(String peerId) {
String peerAddress = getPeerAddress(peerId);
- if(peerAddress != null){
+ if (peerAddress != null) {
return actorSelection(peerAddress);
}
return null;
}
- @Override public void setPeerAddress(String peerId, String peerAddress) {
- Preconditions.checkState(peerAddresses.containsKey(peerId));
- peerAddresses.put(peerId, peerAddress);
- }
-
public void setPeerAddresses(Map<String, String> peerAddresses) {
- this.peerAddresses = peerAddresses;
- }
+ for (String id: getPeerIds()) {
+ removePeer(id);
+ }
- @Override
- public ConfigParams getConfigParams() {
- return configParams;
+ for (Map.Entry<String, String> e: peerAddresses.entrySet()) {
+ addToPeers(e.getKey(), e.getValue(), VotingState.VOTING);
+ }
}
@Override
- public void setSnapshotCaptureInitiated(boolean snapshotCaptureInitiated) {
- this.snapshotCaptureInitiated = snapshotCaptureInitiated;
+ public SnapshotManager getSnapshotManager() {
+ SnapshotManager snapshotManager = super.getSnapshotManager();
+ snapshotManager.setCreateSnapshotRunnable(() -> { });
+ return snapshotManager;
}
@Override
- public boolean isSnapshotCaptureInitiated() {
- return snapshotCaptureInitiated;
+ public RaftPolicy getRaftPolicy() {
+ return raftPolicy != null ? raftPolicy : super.getRaftPolicy();
}
- public void setConfigParams(ConfigParams configParams) {
- this.configParams = configParams;
+ public void setRaftPolicy(RaftPolicy raftPolicy) {
+ this.raftPolicy = raftPolicy;
}
public static class SimpleReplicatedLog extends AbstractReplicatedLogImpl {
- @Override public void appendAndPersist(
+ @Override
+ public int dataSize() {
+ return -1;
+ }
+
+ @Override
+ public void captureSnapshotIfReady(ReplicatedLogEntry replicatedLogEntry) {
+ }
+
+ @Override
+ public boolean removeFromAndPersist(long index) {
+ return removeFrom(index) >= 0;
+ }
+
+ @Override
+ public void appendAndPersist(
ReplicatedLogEntry replicatedLogEntry) {
append(replicatedLogEntry);
}
@Override
- public int dataSize() {
- return -1;
- }
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ public void appendAndPersist(ReplicatedLogEntry replicatedLogEntry, Procedure<ReplicatedLogEntry> callback) {
+ append(replicatedLogEntry);
- @Override public void removeFromAndPersist(long index) {
- removeFrom(index);
+ if (callback != null) {
+ try {
+ callback.apply(replicatedLogEntry);
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ }
}
}
public static class MockPayload extends Payload implements Serializable {
private static final long serialVersionUID = 3121380393130864247L;
private String value = "";
+ private int size;
- public MockPayload(){
-
+ public MockPayload() {
}
- public MockPayload(String s) {
- this.value = s;
+ public MockPayload(String data) {
+ this.value = data;
+ size = value.length();
}
- @Override public Map<GeneratedMessage.GeneratedExtension, String> encode() {
- Map<GeneratedMessage.GeneratedExtension, String> map = new HashMap<GeneratedMessage.GeneratedExtension, String>();
- map.put(MockPayloadMessages.value, value);
- return map;
- }
-
- @Override public Payload decode(
- AppendEntriesMessages.AppendEntries.ReplicatedLogEntry.Payload payloadProtoBuff) {
- String value = payloadProtoBuff.getExtension(MockPayloadMessages.value);
- this.value = value;
- return this;
+ public MockPayload(String data, int size) {
+ this(data);
+ this.size = size;
}
@Override
public int size() {
- return value.length();
- }
-
- @Override public String getClientPayloadClassName() {
- return MockPayload.class.getName();
+ return size;
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
- result = prime * result + ((value == null) ? 0 : value.hashCode());
+ result = prime * result + (value == null ? 0 : value.hashCode());
return result;
}
private final long index;
private final Payload data;
- public MockReplicatedLogEntry(long term, long index, Payload data){
+ public MockReplicatedLogEntry(long term, long index, Payload data) {
this.term = term;
this.index = index;
public int hashCode() {
final int prime = 31;
int result = 1;
- result = prime * result + ((data == null) ? 0 : data.hashCode());
- result = prime * result + (int) (index ^ (index >>> 32));
- result = prime * result + (int) (term ^ (term >>> 32));
+ result = prime * result + (data == null ? 0 : data.hashCode());
+ result = prime * result + (int) (index ^ index >>> 32);
+ result = prime * result + (int) (term ^ term >>> 32);
return result;
}
private final ReplicatedLog mockLog = new SimpleReplicatedLog();
public MockReplicatedLogBuilder createEntries(int start, int end, int term) {
- for (int i=start; i<end; i++) {
- this.mockLog.append(new ReplicatedLogImplEntry(i, term, new MockRaftActorContext.MockPayload("foo" + i)));
+ for (int i = start; i < end; i++) {
+ this.mockLog.append(new ReplicatedLogImplEntry(i, term,
+ new MockRaftActorContext.MockPayload(Integer.toString(i))));
}
return this;
}
return this.mockLog;
}
}
+
+ @Override
+ public void setCurrentBehavior(final RaftActorBehavior behavior) {
+ super.setCurrentBehavior(behavior);
+ }
}