import org.slf4j.LoggerFactory;
public class ClientActor extends UntypedActor {
- protected final Logger LOG = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(ClientActor.class);
private final ActorRef target;
- public ClientActor(ActorRef target){
+ public ClientActor(ActorRef target) {
this.target = target;
}
}
@Override public void onReceive(Object message) throws Exception {
- if(message instanceof KeyValue) {
+ if (message instanceof KeyValue) {
target.tell(message, getSelf());
- } else if(message instanceof KeyValueSaved){
+ } else if (message instanceof KeyValueSaved) {
LOG.info("KeyValue saved");
}
}
import org.opendaylight.yangtools.util.AbstractStringIdentifier;
/**
- * A sample actor showing how the RaftActor is to be extended
+ * A sample actor showing how the RaftActor is to be extended.
*/
public class ExampleActor extends RaftActor implements RaftActorRecoveryCohort, RaftActorSnapshotCohort {
private static final class PayloadIdentifier extends AbstractStringIdentifier<PayloadIdentifier> {
@Override
protected void handleNonRaftCommand(Object message) {
- if(message instanceof KeyValue){
- if(isLeader()) {
+ if (message instanceof KeyValue) {
+ if (isLeader()) {
persistData(getSender(), new PayloadIdentifier(persistIdentifier++), (Payload) message, false);
} else {
- if(getLeader() != null) {
+ if (getLeader() != null) {
getLeader().forward(message, getContext());
}
}
} else if (message instanceof PrintState) {
- if(LOG.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
LOG.debug("State of the node:{} has entries={}, {}",
getId(), state.size(), getReplicatedLogState());
}
} else if (message instanceof PrintRole) {
- if(LOG.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
if (getRaftState() == RaftState.Leader || getRaftState() == RaftState.IsolatedLeader) {
final String followers = ((Leader)this.getCurrentBehavior()).printFollowerStates();
LOG.debug("{} = {}, Peers={}, followers={}", getId(), getRaftState(),
@Override
protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) {
- if(data instanceof KeyValue){
+ if (data instanceof KeyValue) {
KeyValue kv = (KeyValue) data;
state.put(kv.getKey(), kv.getValue());
- if(clientActor != null) {
+ if (clientActor != null) {
clientActor.tell(new KeyValueSaved(), getSelf());
}
}
}
@Override
+ @SuppressWarnings("checkstyle:IllegalCatch")
public void createSnapshot(ActorRef actorRef, java.util.Optional<OutputStream> installSnapshotStream) {
try {
if (installSnapshotStream.isPresent()) {
SerializationUtils.serialize((Serializable) state, installSnapshotStream.get());
}
- } catch (Exception e) {
+ } catch (RuntimeException e) {
LOG.error("Exception in creating snapshot", e);
}
@Override
public void applySnapshot(Snapshot.State snapshotState) {
state.clear();
- try {
- state.putAll(((MapState)snapshotState).state);
- } catch (Exception e) {
- LOG.error("Exception in applying snapshot", e);
- }
- if(LOG.isDebugEnabled()) {
+ state.putAll(((MapState)snapshotState).state);
+
+ if (LOG.isDebugEnabled()) {
LOG.debug("Snapshot applied to state : {}", ((HashMap<?, ?>) state).size());
}
}
import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
/**
- * Implementation of ConfigParams for Example
+ * Implementation of ConfigParams for Example.
*/
public class ExampleConfigParamsImpl extends DefaultConfigParamsImpl {
@Override
import scala.concurrent.duration.FiniteDuration;
/**
- * This is a sample implementation of a Role Change Listener which is an actor, which registers itself to the ClusterRoleChangeNotifier
+ * This is a sample implementation of a Role Change Listener which is an actor, which registers itself
+ * to the ClusterRoleChangeNotifier.
+ *
* <p>
* The Role Change listener receives a SetNotifiers message with the notifiers to register itself with.
+ *
* <p>
- * It kicks of a scheduler which sents registration messages to the notifiers, till it gets a RegisterRoleChangeListenerReply
+ * It kicks of a scheduler which sends registration messages to the notifiers, till it gets a
+ * RegisterRoleChangeListenerReply.
+ *
* <p>
* If all the notifiers have been regsitered with, then it cancels the scheduler.
* It starts the scheduler again when it receives a new registration
- *
*/
-public class ExampleRoleChangeListener extends AbstractUntypedActor implements AutoCloseable{
+public class ExampleRoleChangeListener extends AbstractUntypedActor implements AutoCloseable {
// the akka url should be set to the notifiers actor-system and domain.
private static final String NOTIFIER_AKKA_URL = "akka://raft-test@127.0.0.1:2550/user/";
- private Map<String, Boolean> notifierRegistrationStatus = new HashMap<>();
+ private final Map<String, Boolean> notifierRegistrationStatus = new HashMap<>();
private Cancellable registrationSchedule = null;
- private static final FiniteDuration duration = new FiniteDuration(100, TimeUnit.MILLISECONDS);
- private static final FiniteDuration schedulerDuration = new FiniteDuration(1, TimeUnit.SECONDS);
- private final String memberName;
- private static final String[] shardsToMonitor = new String[] {"example"};
+ private static final FiniteDuration DURATION = new FiniteDuration(100, TimeUnit.MILLISECONDS);
+ private static final FiniteDuration SCHEDULER_DURATION = new FiniteDuration(1, TimeUnit.SECONDS);
+ private static final String[] SHARDS_TO_MONITOR = new String[] {"example"};
public ExampleRoleChangeListener(String memberName) {
- scheduleRegistrationListener(schedulerDuration);
- this.memberName = memberName;
+ scheduleRegistrationListener(SCHEDULER_DURATION);
populateRegistry(memberName);
}
}
private void populateRegistry(String memberName) {
-
- for (String shard: shardsToMonitor) {
- String notifier =(new StringBuilder()).append(NOTIFIER_AKKA_URL).append(memberName)
+ for (String shard: SHARDS_TO_MONITOR) {
+ String notifier = new StringBuilder().append(NOTIFIER_AKKA_URL).append(memberName)
.append("/").append(memberName).append("-notifier").toString();
if (!notifierRegistrationStatus.containsKey(notifier)) {
}
if (!registrationSchedule.isCancelled()) {
- scheduleRegistrationListener(schedulerDuration);
+ scheduleRegistrationListener(SCHEDULER_DURATION);
}
}
+ @SuppressWarnings("checkstyle:IllegalCatch")
private void sendRegistrationRequests() {
for (Map.Entry<String, Boolean> entry : notifierRegistrationStatus.entrySet()) {
if (!entry.getValue()) {
try {
LOG.debug("{} registering with {}", getSelf().path().toString(), entry.getKey());
ActorRef notifier = Await.result(
- getContext().actorSelection(entry.getKey()).resolveOne(duration), duration);
+ getContext().actorSelection(entry.getKey()).resolveOne(DURATION), DURATION);
notifier.tell(new RegisterRoleChangeListener(), getSelf());
}
}
} else {
- LOG.info("Unexpected, RegisterRoleChangeListenerReply received from notifier which is not known to Listener:{}",
+ LOG.info(
+ "Unexpected, RegisterRoleChangeListenerReply received from notifier which is not known to Listener:{}",
senderId);
}
}
package org.opendaylight.controller.cluster.example;
import akka.actor.ActorRef;
-import org.opendaylight.controller.cluster.example.messages.KeyValue;
-
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
+import org.opendaylight.controller.cluster.example.messages.KeyValue;
/**
* Created by kramesha on 7/16/14.
*/
public class LogGenerator {
- private Map<ActorRef, LoggingThread> clientToLoggingThread = new HashMap<>();
+ private
final Map<ActorRef, LoggingThread> clientToLoggingThread = new HashMap<>();
public void startLoggingForClient(ActorRef client) {
LoggingThread lt = new LoggingThread(client);
clientToLoggingThread.put(client, lt);
- Thread t = new Thread(lt);
- t.start();
+ new Thread(lt).start();
}
public void stopLoggingForClient(ActorRef client) {
public class LoggingThread implements Runnable {
- private ActorRef clientActor;
+ private final ActorRef clientActor;
private volatile boolean stopLogging = false;
public LoggingThread(ActorRef clientActor) {
this.clientActor = clientActor;
}
+ @Override
+ @SuppressWarnings("checkstyle:RegexpSingleLineJava")
public void run() {
- Random r = new Random();
+ Random random = new Random();
while (true) {
if (stopLogging) {
System.out.println("Logging stopped for client:" + clientActor.path());
break;
}
String key = clientActor.path().name();
- int random = r.nextInt(100);
- clientActor.tell(new KeyValue(key+"-key-" + random, "value-" + random), null);
+ int randomInt = random.nextInt(100);
+ clientActor.tell(new KeyValue(key + "-key-" + randomInt, "value-" + randomInt), null);
try {
- Thread.sleep((random%10) * 1000);
+ Thread.sleep(randomInt % 10 * 1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
public void startLogging() {
stopLogging = false;
}
-
-
}
-
-
}
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import com.google.common.base.Optional;
-import org.opendaylight.controller.cluster.example.messages.KeyValue;
-import org.opendaylight.controller.cluster.raft.ConfigParams;
-
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.opendaylight.controller.cluster.example.messages.KeyValue;
+import org.opendaylight.controller.cluster.raft.ConfigParams;
-public class Main {
- private static final ActorSystem actorSystem = ActorSystem.create();
+public final class Main {
+ private static final ActorSystem ACTOR_SYSTEM = ActorSystem.create();
// Create three example actors
private static Map<String, String> allPeers = new HashMap<>();
allPeers.put("example-3", "akka://default/user/example-3");
}
- public static void main(String[] args) throws Exception{
+ private Main() {
+ }
+
+ @SuppressWarnings("checkstyle:RegexpSingleLineJava")
+ public static void main(String[] args) throws Exception {
ActorRef example1Actor =
- actorSystem.actorOf(ExampleActor.props("example-1",
+ ACTOR_SYSTEM.actorOf(ExampleActor.props("example-1",
withoutPeer("example-1"), Optional.<ConfigParams>absent()), "example-1");
ActorRef example2Actor =
- actorSystem.actorOf(ExampleActor.props("example-2",
+ ACTOR_SYSTEM.actorOf(ExampleActor.props("example-2",
withoutPeer("example-2"), Optional.<ConfigParams>absent()), "example-2");
ActorRef example3Actor =
- actorSystem.actorOf(ExampleActor.props("example-3",
+ ACTOR_SYSTEM.actorOf(ExampleActor.props("example-3",
withoutPeer("example-3"), Optional.<ConfigParams>absent()), "example-3");
List<ActorRef> examples = Arrays.asList(example1Actor, example2Actor, example3Actor);
- ActorRef clientActor = actorSystem.actorOf(ClientActor.props(example1Actor));
+ ActorRef clientActor = ACTOR_SYSTEM.actorOf(ClientActor.props(example1Actor));
BufferedReader br =
new BufferedReader(new InputStreamReader(System.in));
System.out.println("s <1-3> to start a peer");
System.out.println("k <1-3> to kill a peer");
- while(true) {
+ while (true) {
System.out.print("Enter command (0 to exit):");
try {
- String s = br.readLine();
- String[] split = s.split(" ");
- if(split.length > 1) {
+ String line = br.readLine();
+ String[] split = line.split(" ");
+ if (split.length > 1) {
String command = split[0];
String actor = split[1];
if ("k".equals(command)) {
- int i = Integer.parseInt(actor);
- examples.get(i - 1)
- .tell(PoisonPill.getInstance(), null);
+ int num = Integer.parseInt(actor);
+ examples.get(num - 1).tell(PoisonPill.getInstance(), null);
continue;
} else if ("s".equals(command)) {
- int i = Integer.parseInt(actor);
- String actorName = "example-" + i;
- examples.add(i - 1,
- actorSystem.actorOf(ExampleActor.props(actorName,
+ int num = Integer.parseInt(actor);
+ String actorName = "example-" + num;
+ examples.add(num - 1,
+ ACTOR_SYSTEM.actorOf(ExampleActor.props(actorName,
withoutPeer(actorName), Optional.<ConfigParams>absent()),
actorName));
System.out.println("Created actor : " + actorName);
}
}
- int i = Integer.parseInt(s);
- if(i == 0){
+ int num = Integer.parseInt(line);
+ if (num == 0) {
System.exit(0);
}
- clientActor.tell(new KeyValue("key " + i, "value " + i), null);
+ clientActor.tell(new KeyValue("key " + num, "value " + num), null);
} catch (NumberFormatException nfe) {
System.err.println("Invalid Format!");
}
* Each ExampleActor can have one or more ClientActors. Each ClientActor spawns
* a thread and starts push logs to the actor its assigned to.
*/
+@SuppressWarnings("checkstyle:RegexpSingleLineJava")
public class TestDriver {
-
-
private static Map<String, String> allPeers = new HashMap<>();
private static Map<String, ActorRef> clientActorRefs = new HashMap<>();
private static Map<String, ActorRef> actorRefs = new HashMap<>();
* printNodes
* printState
*
+ * <p>
* Note: when run on IDE and on debug log level, the debug logs in
* AbstractUptypedActor and AbstractUptypedPersistentActor would need to be commented out.
* Also RaftActor handleCommand(), debug log which prints for every command other than AE/AER
- *
- * @param args
- * @throws Exception
*/
public static void main(String[] args) throws Exception {
BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
- while(true) {
+ while (true) {
String command = br.readLine();
if (command.startsWith("bye")) {
System.exit(0);
} else if (command.startsWith("createNodes")) {
String[] arr = command.split(":");
- int n = Integer.parseInt(arr[1]);
- td.createNodes(n);
+ int num = Integer.parseInt(arr[1]);
+ td.createNodes(num);
} else if (command.startsWith("addClients")) {
String[] arr = command.split(":");
- int n = Integer.parseInt(arr[1]);
- td.addClients(n);
+ int num = Integer.parseInt(arr[1]);
+ td.addClients(num);
} else if (command.startsWith("addClientsToNode")) {
String[] arr = command.split(":");
String nodeName = arr[1];
- int n = Integer.parseInt(arr[1]);
- td.addClientsToNode(nodeName, n);
+ int num = Integer.parseInt(arr[1]);
+ td.addClientsToNode(nodeName, num);
} else if (command.startsWith("stopNode")) {
String[] arr = command.split(":");
// create the listener using a separate actor system for each example actor
private static void createClusterRoleChangeListener(List<String> memberIds) {
- System.out.println("memberIds="+memberIds);
+ System.out.println("memberIds=" + memberIds);
for (String memberId : memberIds) {
ActorRef listenerActor = listenerActorSystem.actorOf(
ExampleRoleChangeListener.getProps(memberId), memberId + "-role-change-listener");
}
public void createNodes(int num) {
- for (int i=0; i < num; i++) {
+ for (int i = 0; i < num; i++) {
nameCounter = nameCounter + 1;
- allPeers.put("example-"+nameCounter, "akka://raft-test/user/example-"+nameCounter);
+ allPeers.put("example-" + nameCounter, "akka://raft-test/user/example-" + nameCounter);
}
for (String s : allPeers.keySet()) {
ActorRef exampleActor = createExampleActor(s);
actorRefs.put(s, exampleActor);
- System.out.println("Created node:"+s);
-
+ System.out.println("Created node:" + s);
}
createClusterRoleChangeListener(Lists.newArrayList(allPeers.keySet()));
// add num clients to all nodes in the system
public void addClients(int num) {
- for(Map.Entry<String,ActorRef> actorRefEntry : actorRefs.entrySet()) {
- for (int i=0; i < num; i++) {
+ for (Map.Entry<String, ActorRef> actorRefEntry : actorRefs.entrySet()) {
+ for (int i = 0; i < num; i++) {
String clientName = "client-" + i + "-" + actorRefEntry.getKey();
ActorRef clientActor = actorSystem.actorOf(
ClientActor.props(actorRefEntry.getValue()), clientName);
// add num clients to a node
public void addClientsToNode(String actorName, int num) {
ActorRef actorRef = actorRefs.get(actorName);
- for (int i=0; i < num; i++) {
+ for (int i = 0; i < num; i++) {
String clientName = "client-" + i + "-" + actorName;
clientActorRefs.put(clientName,
actorSystem.actorOf(ClientActor.props(actorRef), clientName));
}
public void reinstateNode(String actorName) {
- String address = "akka://default/user/"+actorName;
+ String address = "akka://default/user/" + actorName;
allPeers.put(actorName, address);
ActorRef exampleActor = createExampleActor(actorName);
}
public void startAllLogging() {
- if(!clientActorRefs.isEmpty()) {
- for(Map.Entry<String,ActorRef> client : clientActorRefs.entrySet()) {
+ if (!clientActorRefs.isEmpty()) {
+ for (Map.Entry<String, ActorRef> client : clientActorRefs.entrySet()) {
logGenerator.startLoggingForClient(client.getValue());
- System.out.println("Started logging for client:"+client.getKey());
+ System.out.println("Started logging for client:" + client.getKey());
}
} else {
- System.out.println("There are no clients for any nodes. First create clients using commands- addClients:<num> or addClientsToNode:<nodename>:<num>");
+ System.out.println(
+ "There are no clients for any nodes. First create clients using commands- addClients:<num> or "
+ + "addClientsToNode:<nodename>:<num>");
}
-
}
public void startLoggingForClient(ActorRef client) {
}
public void stopAllLogging() {
- for(Map.Entry<String,ActorRef> client : clientActorRefs.entrySet()) {
+ for (Map.Entry<String, ActorRef> client : clientActorRefs.entrySet()) {
logGenerator.stopLoggingForClient(client.getValue());
}
}
public KeyValue() {
}
- public KeyValue(String key, String value){
+ public KeyValue(String key, String value) {
this.key = key;
this.value = value;
}
this.value = value;
}
- @Override public String toString() {
- return "KeyValue{" +
- "key='" + key + '\'' +
- ", value='" + value + '\'' +
- '}';
+ @Override
+ public String toString() {
+ return "KeyValue{" + "key='" + key + '\'' + ", value='" + value + '\'' + '}';
}
@Override
package org.opendaylight.controller.cluster.example.messages;
/**
- * Message sent by the Example Role Change Listener to itself for registering itself with the notifiers
- *
- * This message is sent by the scheduler
+ * Message sent by the Example Role Change Listener to itself for registering itself with the notifiers.
+ * This message is sent by the scheduler.
*/
public class RegisterListener {
}
Code Review / controller.git / commitdiff