Integrated Kryo serializers with the communications manager and IO loop stuff.

@@ -29,4 +29,8 @@ public interface ClusterCommunicationAdminService {
      */
     void startUp(DefaultControllerNode localNode, ClusterNodesDelegate delegate);
 
+    /**
+     * Clears all nodes and streams as part of leaving the cluster.
+     */
+    void clearAllNodesAndStreams();
 }

@@ -13,6 +13,7 @@ import org.onlab.onos.cluster.DefaultControllerNode;
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.GoodbyeMessage;
 import org.onlab.onos.store.cluster.messaging.HelloMessage;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.cluster.messaging.MessageSubscriber;
@@ -83,9 +84,11 @@ public class ClusterCommunicationManager
 
     private final Timer timer = new Timer("onos-comm-initiator");
     private final TimerTask connectionCustodian = new ConnectionCustodian();
+    private GoodbyeSubscriber goodbyeSubscriber = new GoodbyeSubscriber();
 
     @Activate
     public void activate() {
+        addSubscriber(MessageSubject.GOODBYE, goodbyeSubscriber);
         log.info("Activated but waiting for delegate");
     }
 
@@ -102,9 +105,20 @@ public class ClusterCommunicationManager
     }
 
     @Override
+    public boolean send(ClusterMessage message) {
+        boolean ok = true;
+        for (DefaultControllerNode node : nodes) {
+            if (!node.equals(localNode)) {
+                ok = send(message, node.id()) && ok;
+            }
+        }
+        return ok;
+    }
+
+    @Override
     public boolean send(ClusterMessage message, NodeId toNodeId) {
         ClusterMessageStream stream = streams.get(toNodeId);
-        if (stream != null) {
+        if (stream != null && !toNodeId.equals(localNode.id())) {
             try {
                 stream.write(message);
                 return true;
@@ -140,6 +154,7 @@ public class ClusterCommunicationManager
 
     @Override
     public void removeNode(DefaultControllerNode node) {
+        send(new GoodbyeMessage(node.id()));
         nodes.remove(node);
         ClusterMessageStream stream = streams.remove(node.id());
         if (stream != null) {
@@ -159,6 +174,16 @@ public class ClusterCommunicationManager
         log.info("Started");
     }
 
+    @Override
+    public void clearAllNodesAndStreams() {
+        nodes.clear();
+        send(new GoodbyeMessage(localNode.id()));
+        for (ClusterMessageStream stream : streams.values()) {
+            stream.close();
+        }
+        streams.clear();
+    }
+
     /**
      * Dispatches the specified message to all subscribers to its subject.
      *
@@ -304,4 +329,11 @@ public class ClusterCommunicationManager
         }
     }
 
+    private class GoodbyeSubscriber implements MessageSubscriber {
+        @Override
+        public void receive(ClusterMessage message, NodeId fromNodeId) {
+            log.info("Received goodbye message from {}", fromNodeId);
+            nodesDelegate.nodeRemoved(fromNodeId);
+        }
+    }
 }

@@ -27,4 +27,11 @@ public interface ClusterNodesDelegate {
      */
     void nodeVanished(NodeId nodeId);
 
+    /**
+     * Notifies about remote request to remove node from cluster.
+     *
+     * @param nodeId identifier of the cluster node that was removed
+     */
+    void nodeRemoved(NodeId nodeId);
+
 }

@@ -127,9 +127,17 @@ public class DistributedClusterStore
 
     @Override
     public void removeNode(NodeId nodeId) {
-        DefaultControllerNode node = nodes.remove(nodeId);
-        if (node != null) {
-            communicationAdminService.removeNode(node);
+        if (nodeId.equals(localNode.id())) {
+            // FIXME: this is still broken
+            // We are being ejected from the cluster, so remove all other nodes.
+            communicationAdminService.clearAllNodesAndStreams();
+            nodes.clear();
+        } else {
+            // Remove the other node.
+            DefaultControllerNode node = nodes.remove(nodeId);
+            if (node != null) {
+                communicationAdminService.removeNode(node);
+            }
         }
     }
 
@@ -148,6 +156,11 @@ public class DistributedClusterStore
         public void nodeVanished(NodeId nodeId) {
             states.put(nodeId, State.INACTIVE);
         }
+
+        @Override
+        public void nodeRemoved(NodeId nodeId) {
+            removeNode(nodeId);
+        }
     }
 
 }

 package org.onlab.onos.store.cluster.impl;
 
+import de.javakaffee.kryoserializers.URISerializer;
+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.onos.cluster.ControllerNode;
+import org.onlab.onos.cluster.DefaultControllerNode;
 import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DefaultDevice;
+import org.onlab.onos.net.DefaultLink;
+import org.onlab.onos.net.DefaultPort;
+import org.onlab.onos.net.Device;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Element;
+import org.onlab.onos.net.Link;
+import org.onlab.onos.net.LinkKey;
+import org.onlab.onos.net.MastershipRole;
+import org.onlab.onos.net.Port;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.provider.ProviderId;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.EchoMessage;
+import org.onlab.onos.store.cluster.messaging.GoodbyeMessage;
 import org.onlab.onos.store.cluster.messaging.HelloMessage;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.cluster.messaging.SerializationService;
+import org.onlab.onos.store.serializers.ConnectPointSerializer;
+import org.onlab.onos.store.serializers.DefaultLinkSerializer;
+import org.onlab.onos.store.serializers.DefaultPortSerializer;
+import org.onlab.onos.store.serializers.DeviceIdSerializer;
+import org.onlab.onos.store.serializers.IpPrefixSerializer;
+import org.onlab.onos.store.serializers.LinkKeySerializer;
+import org.onlab.onos.store.serializers.NodeIdSerializer;
+import org.onlab.onos.store.serializers.PortNumberSerializer;
+import org.onlab.onos.store.serializers.ProviderIdSerializer;
 import org.onlab.packet.IpPrefix;
+import org.onlab.util.KryoPool;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.URI;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
 
 import static com.google.common.base.Preconditions.checkState;
 
@@ -24,11 +57,64 @@ public class MessageSerializer implements SerializationService {
 
     private final Logger log = LoggerFactory.getLogger(getClass());
 
-    private static final int METADATA_LENGTH = 16; // 8 + 4 + 4
-    private static final int LENGTH_OFFSET = 12;
+    private static final int METADATA_LENGTH = 12; // 8 + 4
+    private static final int LENGTH_OFFSET = 8;
 
     private static final long MARKER = 0xfeedcafebeaddeadL;
 
+    private KryoPool serializerPool;
+
+    @Activate
+    public void activate() {
+        setupKryoPool();
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+
+    /**
+     * Sets up the common serialzers pool.
+     */
+    protected void setupKryoPool() {
+        // FIXME Slice out types used in common to separate pool/namespace.
+        serializerPool = KryoPool.newBuilder()
+                .register(ArrayList.class,
+                          HashMap.class,
+
+                          ControllerNode.State.class,
+                          Device.Type.class,
+
+                          DefaultControllerNode.class,
+                          DefaultDevice.class,
+                          MastershipRole.class,
+                          Port.class,
+                          Element.class,
+
+                          Link.Type.class,
+
+                          MessageSubject.class,
+                          HelloMessage.class,
+                          GoodbyeMessage.class,
+                          EchoMessage.class
+                )
+                .register(IpPrefix.class, new IpPrefixSerializer())
+                .register(URI.class, new URISerializer())
+                .register(NodeId.class, new NodeIdSerializer())
+                .register(ProviderId.class, new ProviderIdSerializer())
+                .register(DeviceId.class, new DeviceIdSerializer())
+                .register(PortNumber.class, new PortNumberSerializer())
+                .register(DefaultPort.class, new DefaultPortSerializer())
+                .register(LinkKey.class, new LinkKeySerializer())
+                .register(ConnectPoint.class, new ConnectPointSerializer())
+                .register(DefaultLink.class, new DefaultLinkSerializer())
+                .build()
+                .populate(1);
+    }
+
+
     @Override
     public ClusterMessage decode(ByteBuffer buffer) {
         try {
@@ -47,18 +133,12 @@ public class MessageSerializer implements SerializationService {
             // At this point, we have enough data to read a complete message.
             long marker = buffer.getLong();
             checkState(marker == MARKER, "Incorrect message marker");
-
-            int subjectOrdinal = buffer.getInt();
-            MessageSubject subject = MessageSubject.values()[subjectOrdinal];
             length = buffer.getInt();
 
             // TODO: sanity checking for length
             byte[] data = new byte[length - METADATA_LENGTH];
             buffer.get(data);
-
-            // TODO: add deserialization hook here; for now this hack
-            String[] fields = new String(data).split(":");
-            return new HelloMessage(new NodeId(fields[0]), IpPrefix.valueOf(fields[1]), Integer.parseInt(fields[2]));
+            return (ClusterMessage) serializerPool.deserialize(data);
 
         } catch (Exception e) {
             // TODO: recover from exceptions by forwarding stream to next marker
@@ -70,12 +150,8 @@ public class MessageSerializer implements SerializationService {
     @Override
     public void encode(ClusterMessage message, ByteBuffer buffer) {
         try {
-            HelloMessage helloMessage = (HelloMessage) message;
+            byte[] data = serializerPool.serialize(message);
             buffer.putLong(MARKER);
-            buffer.putInt(message.subject().ordinal());
-
-            String str = helloMessage.nodeId() + ":" + helloMessage.ipAddress() + ":" + helloMessage.tcpPort();
-            byte[] data = str.getBytes();
             buffer.putInt(data.length + METADATA_LENGTH);
             buffer.put(data);
 

@@ -10,6 +10,15 @@ import java.util.Set;
 public interface ClusterCommunicationService {
 
     /**
+     * Sends a message to all controller nodes.
+     *
+     * @param message  message to send
+     * @return true if the message was sent sucessfully to all nodes; false
+     * if there is no stream or if there was an error for some node
+     */
+    boolean send(ClusterMessage message);
+
+    /**
      * Sends a message to the specified controller node.
      *
      * @param message  message to send

+package org.onlab.onos.store.cluster.messaging;
+
+import org.onlab.onos.cluster.NodeId;
+
+/**
+ * Goodbye message that nodes use to leave the cluster for good.
+ */
+public class GoodbyeMessage extends ClusterMessage {
+
+    private NodeId nodeId;
+
+    // For serialization
+    private GoodbyeMessage() {
+        super(MessageSubject.GOODBYE);
+        nodeId = null;
+    }
+
+    /**
+     * Creates a new goodbye message.
+     *
+     * @param nodeId sending node identification
+     */
+    public GoodbyeMessage(NodeId nodeId) {
+        super(MessageSubject.HELLO);
+        this.nodeId = nodeId;
+    }
+
+    /**
+     * Returns the sending node identifer.
+     *
+     * @return node identifier
+     */
+    public NodeId nodeId() {
+        return nodeId;
+    }
+
+}

@@ -8,6 +8,9 @@ public enum MessageSubject {
     /** Represents a first greeting message. */
     HELLO,
 
+    /** Signifies node's intent to leave the cluster. */
+    GOODBYE,
+
     /** Signifies a heart-beat message. */
     ECHO