Preparing for change in ClusterService/Store implementation.
diff --git a/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java b/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java
new file mode 100644
index 0000000..62eed8e
--- /dev/null
+++ b/apps/foo/src/main/java/org/onlab/onos/ccc/DistributedClusterStore.java
@@ -0,0 +1,176 @@
+package org.onlab.onos.ccc;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.nio.AcceptorLoop;
+import org.onlab.nio.IOLoop;
+import org.onlab.nio.MessageStream;
+import org.onlab.onos.cluster.ClusterEvent;
+import org.onlab.onos.cluster.ClusterStore;
+import org.onlab.onos.cluster.ClusterStoreDelegate;
+import org.onlab.onos.cluster.ControllerNode;
+import org.onlab.onos.cluster.DefaultControllerNode;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.store.AbstractStore;
+import org.onlab.packet.IpPrefix;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.ByteChannel;
+import java.nio.channels.ServerSocketChannel;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static java.net.InetAddress.getByAddress;
+import static org.onlab.onos.cluster.ControllerNode.State;
+import static org.onlab.packet.IpPrefix.valueOf;
+import static org.onlab.util.Tools.namedThreads;
+
+/**
+ * Distributed implementation of the cluster nodes store.
+ */
+@Component(immediate = true)
+@Service
+public class DistributedClusterStore
+ extends AbstractStore<ClusterEvent, ClusterStoreDelegate>
+ implements ClusterStore {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private static final long SELECT_TIMEOUT = 50;
+ private static final int WORKERS = 3;
+ private static final int COMM_BUFFER_SIZE = 16 * 1024;
+ private static final int COMM_IDLE_TIME = 500;
+
+ private DefaultControllerNode self;
+ private final Map<NodeId, DefaultControllerNode> nodes = new ConcurrentHashMap<>();
+ private final Map<NodeId, State> states = new ConcurrentHashMap<>();
+
+ private final ExecutorService listenExecutor =
+ Executors.newSingleThreadExecutor(namedThreads("onos-listen"));
+ private final ExecutorService commExecutors =
+ Executors.newFixedThreadPool(WORKERS, namedThreads("onos-cluster"));
+ private final ExecutorService heartbeatExecutor =
+ Executors.newSingleThreadExecutor(namedThreads("onos-heartbeat"));
+
+ private ListenLoop listenLoop;
+ private List<CommLoop> commLoops = new ArrayList<>(WORKERS);
+
+ @Activate
+ public void activate() {
+ establishIdentity();
+ startCommunications();
+ startListening();
+ log.info("Started");
+ }
+
+ private void startCommunications() {
+ for (int i = 0; i < WORKERS; i++) {
+ try {
+ CommLoop loop = new CommLoop();
+ commLoops.add(loop);
+ commExecutors.execute(loop);
+ } catch (IOException e) {
+ log.warn("Unable to start comm IO loop", e);
+ }
+ }
+ }
+
+ // Starts listening for connections from peer cluster members.
+ private void startListening() {
+ try {
+ listenLoop = new ListenLoop(self.ip(), self.tcpPort());
+ listenExecutor.execute(listenLoop);
+ } catch (IOException e) {
+ log.error("Unable to listen for cluster connections", e);
+ }
+ }
+
+ // Establishes the controller's own identity.
+ private void establishIdentity() {
+ // For now rely on env. variable.
+ IpPrefix ip = valueOf(System.getenv("ONOS_NIC"));
+ self = new DefaultControllerNode(new NodeId(ip.toString()), ip);
+ }
+
+ @Deactivate
+ public void deactivate() {
+ listenLoop.shutdown();
+ for (CommLoop loop : commLoops) {
+ loop.shutdown();
+ }
+ log.info("Stopped");
+ }
+
+ @Override
+ public ControllerNode getLocalNode() {
+ return self;
+ }
+
+ @Override
+ public Set<ControllerNode> getNodes() {
+ ImmutableSet.Builder<ControllerNode> builder = ImmutableSet.builder();
+ return builder.addAll(nodes.values()).build();
+ }
+
+ @Override
+ public ControllerNode getNode(NodeId nodeId) {
+ return nodes.get(nodeId);
+ }
+
+ @Override
+ public State getState(NodeId nodeId) {
+ State state = states.get(nodeId);
+ return state == null ? State.INACTIVE : state;
+ }
+
+ @Override
+ public ControllerNode addNode(NodeId nodeId, IpPrefix ip, int tcpPort) {
+ DefaultControllerNode node = new DefaultControllerNode(nodeId, ip, tcpPort);
+ nodes.put(nodeId, node);
+ return node;
+ }
+
+ @Override
+ public void removeNode(NodeId nodeId) {
+ nodes.remove(nodeId);
+ }
+
+ // Listens and accepts inbound connections from other cluster nodes.
+ private class ListenLoop extends AcceptorLoop {
+ ListenLoop(IpPrefix ip, int tcpPort) throws IOException {
+ super(SELECT_TIMEOUT, new InetSocketAddress(getByAddress(ip.toOctets()), tcpPort));
+ }
+
+ @Override
+ protected void acceptConnection(ServerSocketChannel channel) throws IOException {
+
+ }
+ }
+
+ private class CommLoop extends IOLoop<TLVMessage, TLVMessageStream> {
+ CommLoop() throws IOException {
+ super(SELECT_TIMEOUT);
+ }
+
+ @Override
+ protected TLVMessageStream createStream(ByteChannel byteChannel) {
+ return new TLVMessageStream(this, byteChannel, COMM_BUFFER_SIZE, COMM_IDLE_TIME);
+ }
+
+ @Override
+ protected void processMessages(List<TLVMessage> messages, MessageStream<TLVMessage> stream) {
+
+ }
+ }
+}
diff --git a/apps/foo/src/main/java/org/onlab/onos/ccc/TLVMessage.java b/apps/foo/src/main/java/org/onlab/onos/ccc/TLVMessage.java
new file mode 100644
index 0000000..33295a5
--- /dev/null
+++ b/apps/foo/src/main/java/org/onlab/onos/ccc/TLVMessage.java
@@ -0,0 +1,71 @@
+package org.onlab.onos.ccc;
+
+import org.onlab.nio.AbstractMessage;
+
+import java.util.Objects;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+
+/**
+ * Base message for cluster-wide communications using TLVs.
+ */
+public class TLVMessage extends AbstractMessage {
+
+ private final int type;
+ private final Object data;
+
+ /**
+ * Creates an immutable TLV message.
+ *
+ * @param type message type
+ * @param length message length
+ * @param data message data
+ */
+ public TLVMessage(int type, int length, Object data) {
+ this.length = length;
+ this.type = type;
+ this.data = data;
+ }
+
+ /**
+ * Returns the message type indicator.
+ *
+ * @return message type
+ */
+ public int type() {
+ return type;
+ }
+
+ /**
+ * Returns the data object.
+ *
+ * @return message data
+ */
+ public Object data() {
+ return data;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(type, data);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final TLVMessage other = (TLVMessage) obj;
+ return Objects.equals(this.type, other.type) &&
+ Objects.equals(this.data, other.data);
+ }
+
+ @Override
+ public String toString() {
+ return toStringHelper(this).add("type", type).add("length", length).toString();
+ }
+
+}
diff --git a/apps/foo/src/main/java/org/onlab/onos/ccc/TLVMessageStream.java b/apps/foo/src/main/java/org/onlab/onos/ccc/TLVMessageStream.java
new file mode 100644
index 0000000..da0973e
--- /dev/null
+++ b/apps/foo/src/main/java/org/onlab/onos/ccc/TLVMessageStream.java
@@ -0,0 +1,53 @@
+package org.onlab.onos.ccc;
+
+import org.onlab.nio.IOLoop;
+import org.onlab.nio.MessageStream;
+
+import java.nio.ByteBuffer;
+import java.nio.channels.ByteChannel;
+
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Stream for transferring TLV messages between cluster members.
+ */
+public class TLVMessageStream extends MessageStream<TLVMessage> {
+
+ private static final long MARKER = 0xfeedcafecafefeedL;
+
+ /**
+ * Creates a message stream associated with the specified IO loop and
+ * backed by the given byte channel.
+ *
+ * @param loop IO loop
+ * @param byteChannel backing byte channel
+ * @param bufferSize size of the backing byte buffers
+ * @param maxIdleMillis maximum number of millis the stream can be idle
+ */
+ protected TLVMessageStream(IOLoop<TLVMessage, ?> loop, ByteChannel byteChannel,
+ int bufferSize, int maxIdleMillis) {
+ super(loop, byteChannel, bufferSize, maxIdleMillis);
+ }
+
+ @Override
+ protected TLVMessage read(ByteBuffer buffer) {
+ long marker = buffer.getLong();
+ checkState(marker == MARKER, "Incorrect message marker");
+
+ int type = buffer.getInt();
+ int length = buffer.getInt();
+
+ // TODO: add deserialization hook here
+
+ return new TLVMessage(type, length, null);
+ }
+
+ @Override
+ protected void write(TLVMessage message, ByteBuffer buffer) {
+ buffer.putLong(MARKER);
+ buffer.putInt(message.type());
+ buffer.putInt(message.length());
+
+ // TODO: add serialization hook here
+ }
+}