opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleActor.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.example;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.Props;
  13 import com.google.common.base.Optional;
  14 import com.google.protobuf.ByteString;
  15 import java.io.ByteArrayInputStream;
  16 import java.io.ByteArrayOutputStream;
  17 import java.io.IOException;
  18 import java.io.ObjectInputStream;
  19 import java.io.ObjectOutputStream;
  20 import java.util.HashMap;
  21 import java.util.Map;
  22 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  23 import org.opendaylight.controller.cluster.example.messages.KeyValue;
  24 import org.opendaylight.controller.cluster.example.messages.KeyValueSaved;
  25 import org.opendaylight.controller.cluster.example.messages.PrintRole;
  26 import org.opendaylight.controller.cluster.example.messages.PrintState;
  27 import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier;
  28 import org.opendaylight.controller.cluster.raft.ConfigParams;
  29 import org.opendaylight.controller.cluster.raft.RaftActor;
  30 import org.opendaylight.controller.cluster.raft.RaftState;
  31 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
  32 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
  33 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  34
  35 /**
  36  * A sample actor showing how the RaftActor is to be extended
  37  */
  38 public class ExampleActor extends RaftActor {
  39
  40     private final Map<String, String> state = new HashMap();
  41     private final DataPersistenceProvider dataPersistenceProvider;
  42
  43     private long persistIdentifier = 1;
  44     private final Optional<ActorRef> roleChangeNotifier;
  45
  46
  47     public ExampleActor(String id, Map<String, String> peerAddresses,
  48         Optional<ConfigParams> configParams) {
  49         super(id, peerAddresses, configParams);
  50         this.dataPersistenceProvider = new PersistentDataProvider();
  51         roleChangeNotifier = createRoleChangeNotifier(id);
  52     }
  53
  54     public static Props props(final String id, final Map<String, String> peerAddresses,
  55             final Optional<ConfigParams> configParams) {
  56         return Props.create(ExampleActor.class, id, peerAddresses, configParams);
  57     }
  58
  59     @Override public void onReceiveCommand(Object message) throws Exception{
  60         if(message instanceof KeyValue){
  61             if(isLeader()) {
  62                 String persistId = Long.toString(persistIdentifier++);
  63                 persistData(getSender(), persistId, (Payload) message);
  64             } else {
  65                 if(getLeader() != null) {
  66                     getLeader().forward(message, getContext());
  67                 }
  68             }
  69
  70         } else if (message instanceof PrintState) {
  71             if(LOG.isDebugEnabled()) {
  72                 LOG.debug("State of the node:{} has entries={}, {}",
  73                     getId(), state.size(), getReplicatedLogState());
  74             }
  75
  76         } else if (message instanceof PrintRole) {
  77             if(LOG.isDebugEnabled()) {
  78                 if (getRaftState() == RaftState.Leader || getRaftState() == RaftState.IsolatedLeader) {
  79                     final String followers = ((Leader)this.getCurrentBehavior()).printFollowerStates();
  80                     LOG.debug("{} = {}, Peers={}, followers={}", getId(), getRaftState(),
  81                         getRaftActorContext().getPeerAddresses().keySet(), followers);
  82                 } else {
  83                     LOG.debug("{} = {}, Peers={}", getId(), getRaftState(),
  84                         getRaftActorContext().getPeerAddresses().keySet());
  85                 }
  86
  87
  88             }
  89
  90         } else {
  91             super.onReceiveCommand(message);
  92         }
  93     }
  94
  95     protected String getReplicatedLogState() {
  96         return "snapshotIndex=" + getRaftActorContext().getReplicatedLog().getSnapshotIndex()
  97             + ", snapshotTerm=" + getRaftActorContext().getReplicatedLog().getSnapshotTerm()
  98             + ", im-mem journal size=" + getRaftActorContext().getReplicatedLog().size();
  99     }
 100
 101     public Optional<ActorRef> createRoleChangeNotifier(String actorId) {
 102         ActorRef exampleRoleChangeNotifier = this.getContext().actorOf(
 103             RoleChangeNotifier.getProps(actorId), actorId + "-notifier");
 104         return Optional.<ActorRef>of(exampleRoleChangeNotifier);
 105     }
 106
 107     @Override
 108     protected Optional<ActorRef> getRoleChangeNotifier() {
 109         return roleChangeNotifier;
 110     }
 111
 112     @Override protected void applyState(final ActorRef clientActor, final String identifier,
 113         final Object data) {
 114         if(data instanceof KeyValue){
 115             KeyValue kv = (KeyValue) data;
 116             state.put(kv.getKey(), kv.getValue());
 117             if(clientActor != null) {
 118                 clientActor.tell(new KeyValueSaved(), getSelf());
 119             }
 120         }
 121     }
 122
 123     @Override protected void createSnapshot() {
 124         ByteString bs = null;
 125         try {
 126             bs = fromObject(state);
 127         } catch (Exception e) {
 128             LOG.error("Exception in creating snapshot", e);
 129         }
 130         getSelf().tell(new CaptureSnapshotReply(bs.toByteArray()), null);
 131     }
 132
 133     @Override protected void applySnapshot(byte [] snapshot) {
 134         state.clear();
 135         try {
 136             state.putAll((HashMap) toObject(snapshot));
 137         } catch (Exception e) {
 138            LOG.error("Exception in applying snapshot", e);
 139         }
 140         if(LOG.isDebugEnabled()) {
 141             LOG.debug("Snapshot applied to state : {}", ((HashMap) state).size());
 142         }
 143     }
 144
 145     private ByteString fromObject(Object snapshot) throws Exception {
 146         ByteArrayOutputStream b = null;
 147         ObjectOutputStream o = null;
 148         try {
 149             b = new ByteArrayOutputStream();
 150             o = new ObjectOutputStream(b);
 151             o.writeObject(snapshot);
 152             byte[] snapshotBytes = b.toByteArray();
 153             return ByteString.copyFrom(snapshotBytes);
 154         } finally {
 155             if (o != null) {
 156                 o.flush();
 157                 o.close();
 158             }
 159             if (b != null) {
 160                 b.close();
 161             }
 162         }
 163     }
 164
 165     private Object toObject(byte [] bs) throws ClassNotFoundException, IOException {
 166         Object obj = null;
 167         ByteArrayInputStream bis = null;
 168         ObjectInputStream ois = null;
 169         try {
 170             bis = new ByteArrayInputStream(bs);
 171             ois = new ObjectInputStream(bis);
 172             obj = ois.readObject();
 173         } finally {
 174             if (bis != null) {
 175                 bis.close();
 176             }
 177             if (ois != null) {
 178                 ois.close();
 179             }
 180         }
 181         return obj;
 182     }
 183
 184     @Override protected void onStateChanged() {
 185
 186     }
 187
 188     @Override
 189     protected DataPersistenceProvider persistence() {
 190         return dataPersistenceProvider;
 191     }
 192
 193     @Override public void onReceiveRecover(Object message)throws Exception {
 194         super.onReceiveRecover(message);
 195     }
 196
 197     @Override public String persistenceId() {
 198         return getId();
 199     }
 200
 201     @Override
 202     protected void startLogRecoveryBatch(int maxBatchSize) {
 203     }
 204
 205     @Override
 206     protected void appendRecoveredLogEntry(Payload data) {
 207     }
 208
 209     @Override
 210     protected void applyCurrentLogRecoveryBatch() {
 211     }
 212
 213     @Override
 214     protected void onRecoveryComplete() {
 215     }
 216
 217     @Override
 218     protected void applyRecoverySnapshot(byte[] snapshot) {
 219     }
 220 }