X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fexample%2FExampleActor.java;h=06538fd2ae157618a167ef3096f2777981f7bae3;hb=refs%2Fchanges%2F52%2F12352%2F5;hp=aa100df9d0517dfa014a78054e2319d8b1bd34fe;hpb=eb887b1c2c8cd2768f8b4c2ed2b5054f97798466;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleActor.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleActor.java index aa100df9d0..06538fd2ae 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleActor.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleActor.java @@ -12,14 +12,22 @@ import akka.actor.ActorRef; import akka.actor.Props; import akka.japi.Creator; import com.google.common.base.Optional; +import com.google.protobuf.ByteString; +import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.example.messages.KeyValue; import org.opendaylight.controller.cluster.example.messages.KeyValueSaved; import org.opendaylight.controller.cluster.example.messages.PrintRole; import org.opendaylight.controller.cluster.example.messages.PrintState; import org.opendaylight.controller.cluster.raft.ConfigParams; import org.opendaylight.controller.cluster.raft.RaftActor; +import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.util.HashMap; import java.util.Map; @@ -29,6 +37,7 @@ import java.util.Map; public class ExampleActor extends RaftActor { private final Map state = new HashMap(); + private final DataPersistenceProvider dataPersistenceProvider; private long persistIdentifier = 1; @@ -36,6 +45,7 @@ public class ExampleActor extends RaftActor { public ExampleActor(String id, Map peerAddresses, Optional configParams) { super(id, peerAddresses, configParams); + this.dataPersistenceProvider = new PersistentDataProvider(); } public static Props props(final String id, final Map peerAddresses, @@ -48,7 +58,7 @@ public class ExampleActor extends RaftActor { }); } - @Override public void onReceiveCommand(Object message){ + @Override public void onReceiveCommand(Object message) throws Exception{ if(message instanceof KeyValue){ if(isLeader()) { String persistId = Long.toString(persistIdentifier++); @@ -60,11 +70,15 @@ public class ExampleActor extends RaftActor { } } else if (message instanceof PrintState) { - LOG.debug("State of the node:{} has entries={}, {}", - getId(), state.size(), getReplicatedLogState()); + if(LOG.isDebugEnabled()) { + LOG.debug("State of the node:{} has entries={}, {}", + getId(), state.size(), getReplicatedLogState()); + } } else if (message instanceof PrintRole) { - LOG.debug("{} = {}, Peers={}", getId(), getRaftState(),getPeers()); + if(LOG.isDebugEnabled()) { + LOG.debug("{} = {}, Peers={}", getId(), getRaftState(), getPeers()); + } } else { super.onReceiveCommand(message); @@ -82,21 +96,101 @@ public class ExampleActor extends RaftActor { } } - @Override protected Object createSnapshot() { - return state; + @Override protected void createSnapshot() { + ByteString bs = null; + try { + bs = fromObject(state); + } catch (Exception e) { + LOG.error(e, "Exception in creating snapshot"); + } + getSelf().tell(new CaptureSnapshotReply(bs), null); } - @Override protected void applySnapshot(Object snapshot) { + @Override protected void applySnapshot(ByteString snapshot) { state.clear(); - state.putAll((HashMap) snapshot); - LOG.debug("Snapshot applied to state :" + ((HashMap) snapshot).size()); + try { + state.putAll((HashMap) toObject(snapshot)); + } catch (Exception e) { + LOG.error(e, "Exception in applying snapshot"); + } + if(LOG.isDebugEnabled()) { + LOG.debug("Snapshot applied to state : {}", ((HashMap) state).size()); + } + } + + private ByteString fromObject(Object snapshot) throws Exception { + ByteArrayOutputStream b = null; + ObjectOutputStream o = null; + try { + b = new ByteArrayOutputStream(); + o = new ObjectOutputStream(b); + o.writeObject(snapshot); + byte[] snapshotBytes = b.toByteArray(); + return ByteString.copyFrom(snapshotBytes); + } finally { + if (o != null) { + o.flush(); + o.close(); + } + if (b != null) { + b.close(); + } + } + } + + private Object toObject(ByteString bs) throws ClassNotFoundException, IOException { + Object obj = null; + ByteArrayInputStream bis = null; + ObjectInputStream ois = null; + try { + bis = new ByteArrayInputStream(bs.toByteArray()); + ois = new ObjectInputStream(bis); + obj = ois.readObject(); + } finally { + if (bis != null) { + bis.close(); + } + if (ois != null) { + ois.close(); + } + } + return obj; } - @Override public void onReceiveRecover(Object message) { + @Override protected void onStateChanged() { + + } + + @Override + protected DataPersistenceProvider persistence() { + return dataPersistenceProvider; + } + + @Override public void onReceiveRecover(Object message)throws Exception { super.onReceiveRecover(message); } @Override public String persistenceId() { return getId(); } + + @Override + protected void startLogRecoveryBatch(int maxBatchSize) { + } + + @Override + protected void appendRecoveredLogEntry(Payload data) { + } + + @Override + protected void applyCurrentLogRecoveryBatch() { + } + + @Override + protected void onRecoveryComplete() { + } + + @Override + protected void applyRecoverySnapshot(ByteString snapshot) { + } }