ClusterCommunicationManager related serializer fixes.

Change-Id: Id753612ab614b1ee04a64d887a2f7a64b6d9f89a

@@ -29,6 +29,7 @@ import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.serializers.ClusterMessageSerializer;
 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.onos.store.serializers.MessageSubjectSerializer;
 import org.onlab.util.KryoPool;
 import org.onlab.util.KryoPool;
 import org.onlab.netty.Endpoint;
 import org.onlab.netty.Endpoint;
 import org.onlab.netty.Message;
 import org.onlab.netty.Message;
@@ -66,7 +67,7 @@ public class ClusterCommunicationManager
                     .register(ClusterMessage.class, new ClusterMessageSerializer())
                     .register(ClusterMessage.class, new ClusterMessageSerializer())
                     .register(ClusterMembershipEvent.class)
                     .register(ClusterMembershipEvent.class)
                     .register(byte[].class)
                     .register(byte[].class)
-                    .register(MessageSubject.class)
+                    .register(MessageSubject.class, new MessageSubjectSerializer())
                     .build()
                     .build()
                     .populate(1);
                     .populate(1);
         }
         }

@@ -3,7 +3,6 @@ package org.onlab.onos.store.serializers;
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
-
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Input;
@@ -11,6 +10,9 @@ import com.esotericsoftware.kryo.io.Output;
 
 
 public final class ClusterMessageSerializer extends Serializer<ClusterMessage> {
 public final class ClusterMessageSerializer extends Serializer<ClusterMessage> {
 
 
+    /**
+     * Creates a serializer for {@link ClusterMessage}.
+     */
     public ClusterMessageSerializer() {
     public ClusterMessageSerializer() {
         // does not accept null
         // does not accept null
         super(false);
         super(false);

@@ -14,7 +14,7 @@ import com.esotericsoftware.kryo.io.Output;
 public class MastershipBasedTimestampSerializer extends Serializer<MastershipBasedTimestamp> {
 public class MastershipBasedTimestampSerializer extends Serializer<MastershipBasedTimestamp> {
 
 
     /**
     /**
-     * Default constructor.
+     * Creates a serializer for {@link MastershipBasedTimestamp}.
      */
      */
     public MastershipBasedTimestampSerializer() {
     public MastershipBasedTimestampSerializer() {
         // non-null, immutable
         // non-null, immutable

+package org.onlab.onos.store.serializers;
+
+import org.onlab.onos.store.cluster.messaging.MessageSubject;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public final class MessageSubjectSerializer extends Serializer<MessageSubject> {
+
+    /**
+     * Creates a serializer for {@link MessageSubject}.
+     */
+    public MessageSubjectSerializer() {
+        // non-null, immutable
+        super(false, true);
+    }
+
+
+    @Override
+    public void write(Kryo kryo, Output output, MessageSubject object) {
+        output.writeString(object.value());
+    }
+
+    @Override
+    public MessageSubject read(Kryo kryo, Input input,
+            Class<MessageSubject> type) {
+        return new MessageSubject(input.readString());
+    }
+}