- /*
- Here the extended class does whatever it needs to do.
- If it cannot handle a message then it passes it on to the super
- class for handling
- */
- super.onReceiveCommand(message);
+ if(message instanceof KeyValue){
+ if(isLeader()) {
+ String persistId = Long.toString(persistIdentifier++);
+ persistData(getSender(), persistId, (Payload) message);
+ } else {
+ if(getLeader() != null) {
+ getLeader().forward(message, getContext());
+ }
+ }
+
+ } else if (message instanceof PrintState) {
+ LOG.debug("State of the node:"+getId() + " has = "+state.size() + " entries");
+
+ } else if (message instanceof PrintRole) {
+ LOG.debug(getId() + " = " + getRaftState());
+ } else {
+ super.onReceiveCommand(message);
+ }
+ }
+
+ @Override protected void applyState(ActorRef clientActor, String identifier,
+ Object data) {
+ if(data instanceof KeyValue){
+ KeyValue kv = (KeyValue) data;
+ state.put(kv.getKey(), kv.getValue());
+ if(clientActor != null) {
+ clientActor.tell(new KeyValueSaved(), getSelf());
+ }
+ }
+ }
+
+ @Override protected Object createSnapshot() {
+ return state;
+ }
+
+ @Override protected void applySnapshot(Object snapshot) {
+ state.clear();
+ state.putAll((HashMap) snapshot);