Implement generic CustomerSerializers that create more specific AMQPSerializer instances at runtime.

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

@@ -57,6 +57,8 @@ abstract class CustomSerializer<T : Any> : AMQPSerializer<T>, SerializerFor {
     abstract fun writeDescribedObject(obj: T, data: Data, type: Type, output: SerializationOutput,
                                       context: SerializationContext)
 
+    open fun specialiseFor(declaredType: Type): AMQPSerializer<T>? = this
+
     /**
      * This custom serializer represents a sort of symbolic link from a subclass to a super class, where the super
      * class custom serializer is responsible for the "on the wire" format but we want to create a reference to the

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

@@ -129,7 +129,7 @@ class CachingCustomSerializerRegistry(
             customSerializer
         }
     }
-    
+
     override fun findCustomSerializer(clazz: Class<*>, declaredType: Type): AMQPSerializer<Any>? {
         val typeIdentifier = CustomSerializerIdentifier(
                 TypeIdentifier.forClass(clazz),
@@ -173,7 +173,13 @@ class CachingCustomSerializerRegistry(
             throw IllegalCustomSerializerException(declaredSerializers.first(), clazz)
         }
 
-        return declaredSerializers.first()
+        return declaredSerializers.first().let {
+            if (it is CustomSerializer<Any>) {
+                it.specialiseFor(declaredType)
+            } else {
+                it
+            }
+        }
     }
 
     private val Class<*>.isCustomSerializationForbidden: Boolean get() = when {
@@ -182,4 +188,5 @@ class CachingCustomSerializerRegistry(
         isAnnotationPresent(CordaSerializable::class.java) -> true
         else -> false
     }
-}
+}
+

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

@@ -142,12 +142,12 @@ class DeserializationInput constructor(
                 envelope)
     }
 
-    internal fun readObjectOrNull(obj: Any?, schema: SerializationSchemas, type: Type, context: SerializationContext
+    fun readObjectOrNull(obj: Any?, schema: SerializationSchemas, type: Type, context: SerializationContext
     ): Any? {
         return if (obj == null) null else readObject(obj, schema, type, context)
     }
 
-    internal fun readObject(obj: Any, schemas: SerializationSchemas, type: Type, context: SerializationContext): Any =
+    fun readObject(obj: Any, schemas: SerializationSchemas, type: Type, context: SerializationContext): Any =
             if (obj is DescribedType && ReferencedObject.DESCRIPTOR == obj.descriptor) {
                 // It must be a reference to an instance that has already been read, cheaply and quickly returning it by reference.
                 val objectIndex = (obj.described as UnsignedInteger).toInt()