[ONOS-6527] Support serializing multiple types using the same type ID when a serializer is explicitly provided

Change-Id: I4de04eaaea09eb81e2fe8bd28af934170c88a2d8
diff --git a/core/api/src/main/java/org/onosproject/net/resource/DiscreteResourceId.java b/core/api/src/main/java/org/onosproject/net/resource/DiscreteResourceId.java
index 8e51250..441fe88 100644
--- a/core/api/src/main/java/org/onosproject/net/resource/DiscreteResourceId.java
+++ b/core/api/src/main/java/org/onosproject/net/resource/DiscreteResourceId.java
@@ -32,7 +32,7 @@
     private final ImmutableList<Object> components;
 
     DiscreteResourceId(ImmutableList<Object> components) {
-        this.components = ImmutableList.copyOf(components);
+        this.components = components;
     }
 
     DiscreteResourceId() {
diff --git a/core/store/serializers/src/test/java/org/onosproject/store/serializers/KryoSerializerTest.java b/core/store/serializers/src/test/java/org/onosproject/store/serializers/KryoSerializerTest.java
index bfe7815..ed1a473 100644
--- a/core/store/serializers/src/test/java/org/onosproject/store/serializers/KryoSerializerTest.java
+++ b/core/store/serializers/src/test/java/org/onosproject/store/serializers/KryoSerializerTest.java
@@ -127,6 +127,21 @@
     public void tearDown() throws Exception {
     }
 
+    private byte[] serialize(Object object) {
+        ByteBuffer buffer = ByteBuffer.allocate(1024);
+        serializer.encode(object, buffer);
+        buffer.flip();
+        byte[] bytes = new byte[buffer.remaining()];
+        buffer.get(bytes);
+        return bytes;
+    }
+
+    private <T> void testBytesEqual(T expected, T actual) {
+        byte[] expectedBytes = serialize(expected);
+        byte[] actualBytes = serialize(actual);
+        assertArrayEquals(expectedBytes, actualBytes);
+    }
+
     private <T> void testSerializedEquals(T original) {
         ByteBuffer buffer = ByteBuffer.allocate(1 * 1024 * 1024);
         serializer.encode(original, buffer);
@@ -196,6 +211,7 @@
 
     @Test
     public void testImmutableList() {
+        testBytesEqual(ImmutableList.of(DID1, DID2), ImmutableList.of(DID1, DID2, DID1, DID2).subList(0, 2));
         testSerializedEquals(ImmutableList.of(DID1, DID2));
         testSerializedEquals(ImmutableList.of(DID1));
         testSerializedEquals(ImmutableList.of());
diff --git a/utils/misc/src/main/java/org/onlab/util/KryoNamespace.java b/utils/misc/src/main/java/org/onlab/util/KryoNamespace.java
index 7328c8d..1eb1569 100644
--- a/utils/misc/src/main/java/org/onlab/util/KryoNamespace.java
+++ b/utils/misc/src/main/java/org/onlab/util/KryoNamespace.java
@@ -35,6 +35,7 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 
@@ -87,7 +88,7 @@
     public static final class Builder {
 
         private int blockHeadId = INITIAL_ID;
-        private List<Pair<Class<?>, Serializer<?>>> types = new ArrayList<>();
+        private List<Pair<Class<?>[], Serializer<?>>> types = new ArrayList<>();
         private List<RegistrationBlock> blocks = new ArrayList<>();
         private boolean registrationRequired = true;
 
@@ -146,22 +147,23 @@
          */
         public Builder register(final Class<?>... expectedTypes) {
             for (Class<?> clazz : expectedTypes) {
-                types.add(Pair.of(clazz, null));
+                types.add(Pair.of(new Class<?>[]{clazz}, null));
             }
             return this;
         }
 
         /**
-         * Registers a class and it's serializer.
+         * Registers serializer for the given set of classes.
+         * <p>
+         * When multiple classes are registered with an explicitly provided serializer, the namespace guarantees
+         * all instances will be serialized with the same type ID.
          *
          * @param classes list of classes to register
          * @param serializer serializer to use for the class
          * @return this
          */
         public Builder register(Serializer<?> serializer, final Class<?>... classes) {
-            for (Class<?> clazz : classes) {
-                types.add(Pair.of(clazz, serializer));
-            }
+            types.add(Pair.of(classes, checkNotNull(serializer)));
             return this;
         }
 
@@ -429,7 +431,7 @@
             if (id == FLOATING_ID) {
                 id = kryo.getNextRegistrationId();
             }
-            for (Pair<Class<?>, Serializer<?>> entry : block.types()) {
+            for (Pair<Class<?>[], Serializer<?>> entry : block.types()) {
                 register(kryo, entry.getLeft(), entry.getRight(), id++);
             }
         }
@@ -440,36 +442,47 @@
      * Register {@code type} and {@code serializer} to {@code kryo} instance.
      *
      * @param kryo       Kryo instance
-     * @param type       type to register
+     * @param types       types to register
      * @param serializer Specific serializer to register or null to use default.
      * @param id         type registration id to use
      */
-    private void register(Kryo kryo, Class<?> type, Serializer<?> serializer, int id) {
+    private void register(Kryo kryo, Class<?>[] types, Serializer<?> serializer, int id) {
         Registration existing = kryo.getRegistration(id);
         if (existing != null) {
-            if (existing.getType() != type) {
+            boolean matches = false;
+            for (Class<?> type : types) {
+                if (existing.getType() == type) {
+                    matches = true;
+                    break;
+                }
+            }
+
+            if (!matches) {
                 log.error("{}: Failed to register {} as {}, {} was already registered.",
-                          friendlyName(), type, id, existing.getType());
+                          friendlyName(), types, id, existing.getType());
 
                 throw new IllegalStateException(String.format(
                           "Failed to register %s as %s, %s was already registered.",
-                          type, id, existing.getType()));
+                          Arrays.toString(types), id, existing.getType()));
             }
             // falling through to register call for now.
             // Consider skipping, if there's reasonable
             // way to compare serializer equivalence.
         }
-        Registration r;
-        if (serializer == null) {
-            r = kryo.register(type, id);
-        } else {
-            r = kryo.register(type, serializer, id);
+
+        for (Class<?> type : types) {
+            Registration r;
+            if (serializer == null) {
+                r = kryo.register(type, id);
+            } else {
+                r = kryo.register(type, serializer, id);
+            }
+            if (r.getId() != id) {
+                log.warn("{}: {} already registed as {}. Skipping {}.",
+                        friendlyName(), r.getType(), r.getId(), id);
+            }
+            log.trace("{} registered as {}", r.getType(), r.getId());
         }
-        if (r.getId() != id) {
-            log.warn("{}: {} already registed as {}. Skipping {}.",
-                     friendlyName(), r.getType(), r.getId(), id);
-        }
-        log.trace("{} registered as {}", r.getType(), r.getId());
     }
 
     @Override
@@ -503,9 +516,9 @@
 
     static final class RegistrationBlock {
         private final int begin;
-        private final ImmutableList<Pair<Class<?>, Serializer<?>>> types;
+        private final ImmutableList<Pair<Class<?>[], Serializer<?>>> types;
 
-        public RegistrationBlock(int begin, List<Pair<Class<?>, Serializer<?>>> types) {
+        public RegistrationBlock(int begin, List<Pair<Class<?>[], Serializer<?>>> types) {
             this.begin = begin;
             this.types = ImmutableList.copyOf(types);
         }
@@ -514,7 +527,7 @@
             return begin;
         }
 
-        public ImmutableList<Pair<Class<?>, Serializer<?>>> types() {
+        public ImmutableList<Pair<Class<?>[], Serializer<?>>> types() {
             return types;
         }