X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fmessages%2FInstallSnapshot.java;h=d2a6a0d0824b53895e860491c7958aa50132c65d;hp=5a0cc95c21964168f305445d129b9dfe3f7d6bbb;hb=9d5ec5cdd146a56bc03e35b6718e9492a5c8410a;hpb=b17a51ecb983331f0e521e40f9dd2474f268de13 diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/InstallSnapshot.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/InstallSnapshot.java index 5a0cc95c21..d2a6a0d082 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/InstallSnapshot.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/InstallSnapshot.java @@ -8,22 +8,37 @@ package org.opendaylight.controller.cluster.raft.messages; +import com.google.common.base.Optional; import com.google.protobuf.ByteString; +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import org.opendaylight.controller.cluster.raft.RaftVersions; +import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload; import org.opendaylight.controller.protobuff.messages.cluster.raft.InstallSnapshotMessages; -public class InstallSnapshot extends AbstractRaftRPC { - +public class InstallSnapshot extends AbstractRaftRPC implements Externalizable { + private static final long serialVersionUID = 1L; public static final Class SERIALIZABLE_CLASS = InstallSnapshotMessages.InstallSnapshot.class; - private final String leaderId; - private final long lastIncludedIndex; - private final long lastIncludedTerm; - private final ByteString data; - private final int chunkIndex; - private final int totalChunks; + private String leaderId; + private long lastIncludedIndex; + private long lastIncludedTerm; + private byte[] data; + private int chunkIndex; + private int totalChunks; + private Optional lastChunkHashCode; + private Optional serverConfig; + + /** + * Empty constructor to satisfy Externalizable. + */ + public InstallSnapshot() { + } - public InstallSnapshot(long term, String leaderId, long lastIncludedIndex, - long lastIncludedTerm, ByteString data, int chunkIndex, int totalChunks) { + public InstallSnapshot(long term, String leaderId, long lastIncludedIndex, long lastIncludedTerm, byte[] data, + int chunkIndex, int totalChunks, Optional lastChunkHashCode, Optional serverConfig) { super(term); this.leaderId = leaderId; this.lastIncludedIndex = lastIncludedIndex; @@ -31,6 +46,14 @@ public class InstallSnapshot extends AbstractRaftRPC { this.data = data; this.chunkIndex = chunkIndex; this.totalChunks = totalChunks; + this.lastChunkHashCode = lastChunkHashCode; + this.serverConfig = serverConfig; + } + + public InstallSnapshot(long term, String leaderId, long lastIncludedIndex, + long lastIncludedTerm, byte[] data, int chunkIndex, int totalChunks) { + this(term, leaderId, lastIncludedIndex, lastIncludedTerm, data, chunkIndex, totalChunks, + Optional.absent(), Optional.absent()); } public String getLeaderId() { @@ -45,7 +68,7 @@ public class InstallSnapshot extends AbstractRaftRPC { return lastIncludedTerm; } - public ByteString getData() { + public byte[] getData() { return data; } @@ -57,26 +80,113 @@ public class InstallSnapshot extends AbstractRaftRPC { return totalChunks; } - public Object toSerializable(){ - return InstallSnapshotMessages.InstallSnapshot.newBuilder() - .setLeaderId(this.getLeaderId()) - .setChunkIndex(this.getChunkIndex()) - .setData(this.getData()) - .setLastIncludedIndex(this.getLastIncludedIndex()) - .setLastIncludedTerm(this.getLastIncludedTerm()) - .setTotalChunks(this.getTotalChunks()).build(); + public Optional getLastChunkHashCode() { + return lastChunkHashCode; + } + public Optional getServerConfig() { + return serverConfig; } - public static InstallSnapshot fromSerializable (Object o) { - InstallSnapshotMessages.InstallSnapshot from = - (InstallSnapshotMessages.InstallSnapshot) o; + @Override + public void writeExternal(ObjectOutput out) throws IOException { + out.writeShort(RaftVersions.CURRENT_VERSION); + out.writeLong(getTerm()); + out.writeUTF(leaderId); + out.writeLong(lastIncludedIndex); + out.writeLong(lastIncludedTerm); + out.writeInt(chunkIndex); + out.writeInt(totalChunks); + + out.writeByte(lastChunkHashCode.isPresent() ? 1 : 0); + if(lastChunkHashCode.isPresent()) { + out.writeInt(lastChunkHashCode.get().intValue()); + } + + out.writeByte(serverConfig.isPresent() ? 1 : 0); + if(serverConfig.isPresent()) { + out.writeObject(serverConfig.get()); + } + + out.writeObject(data); + } - InstallSnapshot installSnapshot = new InstallSnapshot(from.getTerm(), - from.getLeaderId(), from.getLastIncludedIndex(), - from.getLastIncludedTerm(), from.getData(), - from.getChunkIndex(), from.getTotalChunks()); + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + in.readShort(); // raft version - not currently used + setTerm(in.readLong()); + leaderId = in.readUTF(); + lastIncludedIndex = in.readLong(); + lastIncludedTerm = in.readLong(); + chunkIndex = in.readInt(); + totalChunks = in.readInt(); + + lastChunkHashCode = Optional.absent(); + boolean chunkHashCodePresent = in.readByte() == 1; + if(chunkHashCodePresent) { + lastChunkHashCode = Optional.of(in.readInt()); + } + + serverConfig = Optional.absent(); + boolean serverConfigPresent = in.readByte() == 1; + if(serverConfigPresent) { + serverConfig = Optional.of((ServerConfigurationPayload)in.readObject()); + } + + data = (byte[])in.readObject(); + } + + public Object toSerializable(short version) { + if(version >= RaftVersions.BORON_VERSION) { + return this; + } else { + InstallSnapshotMessages.InstallSnapshot.Builder builder = InstallSnapshotMessages.InstallSnapshot.newBuilder() + .setTerm(this.getTerm()) + .setLeaderId(this.getLeaderId()) + .setChunkIndex(this.getChunkIndex()) + .setData(ByteString.copyFrom(getData())) + .setLastIncludedIndex(this.getLastIncludedIndex()) + .setLastIncludedTerm(this.getLastIncludedTerm()) + .setTotalChunks(this.getTotalChunks()); + + if(lastChunkHashCode.isPresent()){ + builder.setLastChunkHashCode(lastChunkHashCode.get()); + } + return builder.build(); + } + } + + @Override + public String toString() { + return "InstallSnapshot [term=" + getTerm() + ", leaderId=" + leaderId + ", lastIncludedIndex=" + + lastIncludedIndex + ", lastIncludedTerm=" + lastIncludedTerm + ", datasize=" + data.length + + ", Chunk=" + chunkIndex + "/" + totalChunks + ", lastChunkHashCode=" + lastChunkHashCode + + ", serverConfig=" + serverConfig.orNull() + "]"; + } + + public static InstallSnapshot fromSerializable (Object o) { + if(o instanceof InstallSnapshot) { + return (InstallSnapshot)o; + } else { + InstallSnapshotMessages.InstallSnapshot from = + (InstallSnapshotMessages.InstallSnapshot) o; + + Optional lastChunkHashCode = Optional.absent(); + if(from.hasLastChunkHashCode()){ + lastChunkHashCode = Optional.of(from.getLastChunkHashCode()); + } + + InstallSnapshot installSnapshot = new InstallSnapshot(from.getTerm(), + from.getLeaderId(), from.getLastIncludedIndex(), + from.getLastIncludedTerm(), from.getData().toByteArray(), + from.getChunkIndex(), from.getTotalChunks(), lastChunkHashCode, + Optional.absent()); + + return installSnapshot; + } + } - return installSnapshot; + public static boolean isSerializedType(Object message) { + return message instanceof InstallSnapshot || message instanceof InstallSnapshotMessages.InstallSnapshot; } }