#224 correcting ordering and adding tests units

kafka/src/main/java/io/micronaut/configuration/kafka/serde/CompositeSerdeRegistry.java

@@ -49,7 +49,7 @@ public CompositeSerdeRegistry(SerdeRegistry... registries) {
                 registries != null ? Arrays.asList(registries) : Collections.emptyList()
         );
 
-        OrderUtil.sort(registries);
+        OrderUtil.sort(this.registries);
     }
 
     @SuppressWarnings("unchecked")

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/CompositeSerdeRegistrySpec.groovy

@@ -0,0 +1,51 @@
+package io.micronaut.configuration.kafka.serde
+
+import io.micronaut.configuration.kafka.serde.serderegistry.Pojo2SerdeRegistryDefaultPrecedence
+import io.micronaut.configuration.kafka.serde.serderegistry.Pojo1
+import io.micronaut.configuration.kafka.serde.serderegistry.Pojo1SerdeRegistryDefaultPrecedence
+import io.micronaut.configuration.kafka.serde.serderegistry.Pojo1SerdeRegistryHighPrededence
+import io.micronaut.configuration.kafka.serde.serderegistry.Pojo1SerdeRegistryLowestPrecedence
+import org.apache.kafka.common.serialization.Serde
+import spock.lang.Specification
+
+class CompositeSerdeRegistrySpec extends Specification {
+
+    void "test highest precedence serde"() {
+        setup:
+        CompositeSerdeRegistry registry = new CompositeSerdeRegistry(
+                new Pojo2SerdeRegistryDefaultPrecedence(),
+                new Pojo1SerdeRegistryDefaultPrecedence(),
+                new Pojo1SerdeRegistryLowestPrecedence(),
+                new Pojo1SerdeRegistryHighPrededence()
+        )
+        when:
+        Serde<?> serde = registry.getSerde(Pojo1)
+        then:
+        serde.class == Pojo1SerdeRegistryHighPrededence.StringSerde
+    }
+
+    void "test default precedence serde"() {
+        setup:
+        CompositeSerdeRegistry registry = new CompositeSerdeRegistry(
+                new Pojo2SerdeRegistryDefaultPrecedence(),
+                new Pojo1SerdeRegistryDefaultPrecedence(),
+                new Pojo1SerdeRegistryLowestPrecedence()
+        )
+        when:
+        Serde<?> serde = registry.getSerde(Pojo1)
+        then:
+        serde.class == Pojo1SerdeRegistryDefaultPrecedence.StringSerde
+    }
+
+    void "test lowest precedence serde"() {
+        setup:
+        CompositeSerdeRegistry registry = new CompositeSerdeRegistry(
+                new Pojo2SerdeRegistryDefaultPrecedence(),
+                new Pojo1SerdeRegistryLowestPrecedence()
+        )
+        when:
+        Serde<?> serde = registry.getSerde(Pojo1)
+        then:
+        serde.class == Pojo1SerdeRegistryLowestPrecedence.StringSerde
+    }
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/serderegistry/Pojo1.groovy

@@ -0,0 +1,4 @@
+package io.micronaut.configuration.kafka.serde.serderegistry
+
+class Pojo1 {
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/serderegistry/Pojo1SerdeRegistryDefaultPrecedence.groovy

@@ -0,0 +1,26 @@
+package io.micronaut.configuration.kafka.serde.serderegistry
+
+import groovy.transform.CompileStatic
+import io.micronaut.configuration.kafka.serde.SerdeRegistry
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+
+@CompileStatic
+class Pojo1SerdeRegistryDefaultPrecedence implements SerdeRegistry {
+
+    static class StringSerde extends Serdes.WrapperSerde<String> {
+        StringSerde() {
+            super(new StringSerializer(), new StringDeserializer())
+        }
+    }
+
+    @Override
+    <T> Serde<T> getSerde(Class<T> type) {
+        if (type == Pojo1) {
+            return new StringSerde() as Serde<T>
+        }
+        return null
+    }
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/serderegistry/Pojo1SerdeRegistryHighPrededence.groovy

@@ -0,0 +1,29 @@
+package io.micronaut.configuration.kafka.serde.serderegistry
+
+import groovy.transform.CompileStatic
+import io.micronaut.configuration.kafka.serde.SerdeRegistry
+import io.micronaut.core.order.Ordered
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+
+@CompileStatic
+class Pojo1SerdeRegistryHighPrededence implements SerdeRegistry, Ordered {
+
+    int order = HIGHEST_PRECEDENCE
+
+    static class StringSerde extends Serdes.WrapperSerde<String> {
+        StringSerde() {
+            super(new StringSerializer(), new StringDeserializer())
+        }
+    }
+
+    @Override
+    <T> Serde<T> getSerde(Class<T> type) {
+        if (type == Pojo1) {
+            return new StringSerde() as Serde<T>
+        }
+        return null
+    }
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/serderegistry/Pojo1SerdeRegistryLowestPrecedence.groovy

@@ -0,0 +1,29 @@
+package io.micronaut.configuration.kafka.serde.serderegistry
+
+import groovy.transform.CompileStatic
+import io.micronaut.configuration.kafka.serde.SerdeRegistry
+import io.micronaut.core.order.Ordered
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+
+@CompileStatic
+class Pojo1SerdeRegistryLowestPrecedence implements SerdeRegistry, Ordered {
+
+    int order = LOWEST_PRECEDENCE
+
+    static class StringSerde extends Serdes.WrapperSerde<String> {
+        StringSerde() {
+            super(new StringSerializer(), new StringDeserializer())
+        }
+    }
+
+    @Override
+    <T> Serde<T> getSerde(Class<T> type) {
+        if (type == Pojo1) {
+            return new StringSerde() as Serde<T>
+        }
+        return null
+    }
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/serderegistry/Pojo2.groovy

@@ -0,0 +1,4 @@
+package io.micronaut.configuration.kafka.serde.serderegistry
+
+class Pojo2 {
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/serde/serderegistry/Pojo2SerdeRegistryDefaultPrecedence.groovy

@@ -0,0 +1,25 @@
+package io.micronaut.configuration.kafka.serde.serderegistry
+
+import groovy.transform.CompileStatic
+import io.micronaut.configuration.kafka.serde.SerdeRegistry
+import org.apache.kafka.common.serialization.IntegerDeserializer
+import org.apache.kafka.common.serialization.IntegerSerializer
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serdes
+
+@CompileStatic
+class Pojo2SerdeRegistryDefaultPrecedence implements SerdeRegistry {
+
+    static class IntegerSerde extends Serdes.WrapperSerde<Integer> {
+        IntegerSerde() {
+            super(new IntegerSerializer(), new IntegerDeserializer())
+        }
+    }
+    @Override
+    <T> Serde<T> getSerde(Class<T> type) {
+        if (type == Pojo2) {
+            return new IntegerSerde() as Serde<T>
+        }
+        return null
+    }
+}