ClusterCommunicationManager bugfixes

Change-Id: I0cb433bf2197c7c745733657607d9e62bb23567d

@@ -26,6 +26,7 @@ import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
 import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
+import org.onlab.onos.store.serializers.ClusterMessageSerializer;
 import org.onlab.onos.store.serializers.KryoPoolUtil;
 import org.onlab.onos.store.serializers.KryoPoolUtil;
 import org.onlab.onos.store.serializers.KryoSerializer;
 import org.onlab.onos.store.serializers.KryoSerializer;
 import org.onlab.util.KryoPool;
 import org.onlab.util.KryoPool;
@@ -50,8 +51,6 @@ public class ClusterCommunicationManager
     private ClusterService clusterService;
     private ClusterService clusterService;
 
 
     private ClusterNodesDelegate nodesDelegate;
     private ClusterNodesDelegate nodesDelegate;
-    // FIXME: `members` should go away and should be using ClusterService
-    private Map<NodeId, ControllerNode> members = new HashMap<>();
     private final Timer timer = new Timer("onos-controller-heatbeats");
     private final Timer timer = new Timer("onos-controller-heatbeats");
     public static final long HEART_BEAT_INTERVAL_MILLIS = 1000L;
     public static final long HEART_BEAT_INTERVAL_MILLIS = 1000L;
 
 
@@ -59,11 +58,14 @@ public class ClusterCommunicationManager
     private MessagingService messagingService;
     private MessagingService messagingService;
 
 
     private static final KryoSerializer SERIALIZER = new KryoSerializer() {
     private static final KryoSerializer SERIALIZER = new KryoSerializer() {
+        @Override
         protected void setupKryoPool() {
         protected void setupKryoPool() {
             serializerPool = KryoPool.newBuilder()
             serializerPool = KryoPool.newBuilder()
                     .register(KryoPoolUtil.API)
                     .register(KryoPoolUtil.API)
-                    .register(ClusterMessage.class)
+                    .register(ClusterMessage.class, new ClusterMessageSerializer())
                     .register(ClusterMembershipEvent.class)
                     .register(ClusterMembershipEvent.class)
+                    .register(byte[].class)
+                    .register(MessageSubject.class)
                     .build()
                     .build()
                     .populate(1);
                     .populate(1);
         }
         }
@@ -73,7 +75,15 @@ public class ClusterCommunicationManager
     @Activate
     @Activate
     public void activate() {
     public void activate() {
         localNode = clusterService.getLocalNode();
         localNode = clusterService.getLocalNode();
-        messagingService = new NettyMessagingService(localNode.tcpPort());
+        NettyMessagingService netty = new NettyMessagingService(localNode.tcpPort());
+        // FIXME: workaround until it becomes a service.
+        try {
+            netty.activate();
+        } catch (Exception e) {
+            // TODO Auto-generated catch block
+            log.error("NettyMessagingService#activate", e);
+        }
+        messagingService = netty;
         log.info("Started");
         log.info("Started");
     }
     }
 
 
@@ -86,7 +96,7 @@ public class ClusterCommunicationManager
     @Override
     @Override
     public boolean broadcast(ClusterMessage message) {
     public boolean broadcast(ClusterMessage message) {
         boolean ok = true;
         boolean ok = true;
-        for (ControllerNode node : members.values()) {
+        for (ControllerNode node : clusterService.getNodes()) {
             if (!node.equals(localNode)) {
             if (!node.equals(localNode)) {
                 ok = unicast(message, node.id()) && ok;
                 ok = unicast(message, node.id()) && ok;
             }
             }
@@ -107,7 +117,7 @@ public class ClusterCommunicationManager
 
 
     @Override
     @Override
     public boolean unicast(ClusterMessage message, NodeId toNodeId) {
     public boolean unicast(ClusterMessage message, NodeId toNodeId) {
-        ControllerNode node = members.get(toNodeId);
+        ControllerNode node = clusterService.getNode(toNodeId);
         checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
         checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
         Endpoint nodeEp = new Endpoint(node.ip().toString(), node.tcpPort());
         Endpoint nodeEp = new Endpoint(node.ip().toString(), node.tcpPort());
         try {
         try {
@@ -137,7 +147,7 @@ public class ClusterCommunicationManager
 
 
     @Override
     @Override
     public void addNode(ControllerNode node) {
     public void addNode(ControllerNode node) {
-        members.put(node.id(), node);
+        //members.put(node.id(), node);
     }
     }
 
 
     @Override
     @Override
@@ -146,7 +156,7 @@ public class ClusterCommunicationManager
                 localNode.id(),
                 localNode.id(),
                 new MessageSubject("CLUSTER_MEMBERSHIP_EVENT"),
                 new MessageSubject("CLUSTER_MEMBERSHIP_EVENT"),
                 SERIALIZER.encode(new ClusterMembershipEvent(ClusterMembershipEventType.LEAVING_MEMBER, node))));
                 SERIALIZER.encode(new ClusterMembershipEvent(ClusterMembershipEventType.LEAVING_MEMBER, node))));
-        members.remove(node.id());
+        //members.remove(node.id());
     }
     }
 
 
     // Sends a heart beat to all peers.
     // Sends a heart beat to all peers.