CORDA-3218: Make set of serializer types considered suitable for object reference to be configurable.

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/DeserializationInput.kt

@@ -188,7 +188,7 @@ class DeserializationInput constructor(
                 // Store the reference in case we need it later on.
                 // Skip for primitive types as they are too small and overhead of referencing them will be much higher
                 // than their content
-                if (suitableForObjectReference(objectRead.javaClass)) {
+                if (serializerFactory.isSuitebleForObjectReference(objectRead.javaClass)) {
                     objectHistory.add(objectRead)
                 }
                 objectRead

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/LocalSerializerFactory.kt

@@ -13,6 +13,7 @@ import java.lang.reflect.ParameterizedType
 import java.lang.reflect.Type
 import java.util.*
 import java.util.function.Function
+import java.util.function.Predicate
 import javax.annotation.concurrent.ThreadSafe
 
 /**
@@ -77,6 +78,12 @@ interface LocalSerializerFactory {
      * Use the [FingerPrinter] to create a type descriptor for the given [typeInformation].
      */
     fun createDescriptor(typeInformation: LocalTypeInformation): Symbol
+
+    /**
+     * Determines whether instances of this type should be added to the object history
+     * when serialising and deserialising.
+     */
+    fun isSuitebleForObjectReference(type: Type): Boolean
 }
 
 /**
@@ -91,6 +98,7 @@ class DefaultLocalSerializerFactory(
         override val classloader: ClassLoader,
         private val descriptorBasedSerializerRegistry: DescriptorBasedSerializerRegistry,
         private val primitiveSerializerFactory: Function<Class<*>, AMQPSerializer<Any>>,
+        private val isPrimitiveType: Predicate<Class<*>>,
         private val customSerializerRegistry: CustomSerializerRegistry,
         private val onlyCustomSerializers: Boolean)
     : LocalSerializerFactory {
@@ -127,6 +135,12 @@ class DefaultLocalSerializerFactory(
     override fun get(typeInformation: LocalTypeInformation): AMQPSerializer<Any> =
             get(typeInformation.observedType, typeInformation)
 
+    // ByteArrays, primitives and boxed primitives are not stored in the object history
+    override fun isSuitebleForObjectReference(type: Type): Boolean {
+        val clazz = type.asClass()
+        return type != ByteArray::class.java && !isPrimitiveType.test(clazz)
+    }
+
     private fun makeAndCache(typeInformation: LocalTypeInformation, build: () -> AMQPSerializer<Any>) =
             makeAndCache(typeInformation.typeIdentifier, build)
 
@@ -144,7 +158,9 @@ class DefaultLocalSerializerFactory(
             // Any Custom Serializer cached for a ParameterizedType can only be
             // found by searching for that exact same type. Searching for its raw
             // class will not work!
-            val declaredGenericType = if (declaredType !is ParameterizedType && localTypeInformation.typeIdentifier is Parameterised) {
+            val declaredGenericType = if (declaredType !is ParameterizedType
+                    && localTypeInformation.typeIdentifier is Parameterised
+                    && declaredClass != Class::class.java) {
                 localTypeInformation.typeIdentifier.getLocalType(classLoaderFor(declaredClass))
             } else {
                 declaredType

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/SerializationHelper.kt

@@ -1,6 +1,5 @@
 package net.corda.serialization.internal.amqp
 
-import com.google.common.primitives.Primitives
 import com.google.common.reflect.TypeToken
 import net.corda.core.serialization.*
 import net.corda.serialization.internal.model.TypeIdentifier
@@ -109,12 +108,6 @@ internal fun Type.isSubClassOf(type: Type): Boolean {
     return TypeToken.of(this).isSubtypeOf(TypeToken.of(type).rawType)
 }
 
-// ByteArrays, primitives and boxed primitives are not stored in the object history
-internal fun suitableForObjectReference(type: Type): Boolean {
-    val clazz = type.asClass()
-    return type != ByteArray::class.java && (!clazz.isPrimitive && !Primitives.unwrap(clazz).isPrimitive)
-}
-
 /**
  * Common properties that are to be used in the [SerializationContext.properties] to alter serialization behavior/content
  */

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/SerializationOutput.kt

@@ -137,7 +137,7 @@ open class SerializationOutput constructor(
             // Important to do it after serialization such that dependent object will have preceding reference numbers
             // assigned to them first as they will be first read from the stream on receiving end.
             // Skip for primitive types as they are too small and overhead of referencing them will be much higher than their content
-            if (suitableForObjectReference(obj.javaClass)) {
+            if (serializerFactory.isSuitebleForObjectReference(obj.javaClass)) {
                 objectHistory[obj] = objectHistory.size
             }
         } else {

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/SerializerFactoryBuilder.kt

@@ -1,5 +1,6 @@
 package net.corda.serialization.internal.amqp
 
+import com.google.common.primitives.Primitives
 import net.corda.core.DeleteForDJVM
 import net.corda.core.KeepForDJVM
 import net.corda.core.serialization.ClassWhitelist
@@ -9,6 +10,7 @@ import net.corda.serialization.internal.model.*
 import java.io.NotSerializableException
 import java.util.Collections.unmodifiableMap
 import java.util.function.Function
+import java.util.function.Predicate
 
 @KeepForDJVM
 object SerializerFactoryBuilder {
@@ -109,6 +111,7 @@ object SerializerFactoryBuilder {
                 classCarpenter.classloader,
                 descriptorBasedSerializerRegistry,
                 Function { clazz -> AMQPPrimitiveSerializer(clazz) },
+                Predicate { clazz -> clazz.isPrimitive || Primitives.unwrap(clazz).isPrimitive },
                 customSerializerRegistry,
                 onlyCustomSerializers)
 

serialization/src/main/kotlin/net/corda/serialization/internal/model/TypeModellingFingerPrinter.kt

@@ -236,7 +236,9 @@ private class FingerPrintingState(
         // Any Custom Serializer cached for a ParameterizedType can only be
         // found by searching for that exact same type. Searching for its raw
         // class will not work!
-        val observedGenericType = if (observedType !is ParameterizedType && type.typeIdentifier is Parameterised) {
+        val observedGenericType = if (observedType !is ParameterizedType
+                && type.typeIdentifier is Parameterised
+                && observedClass != Class::class.java) {
             type.typeIdentifier.getLocalType(classLoaderFor(observedClass))
         } else {
             observedType