Merge pull request #2017 from corda/kat/feature/enableAMQP

CORDA-780 - Enable AMQP for P2P and Storage Contexts

.ci/api-current.txt

@@ -2724,6 +2724,10 @@ public static final class net.corda.core.serialization.SerializationContext$UseC
   public static net.corda.core.serialization.SerializationContext$UseCase valueOf(String)
   public static net.corda.core.serialization.SerializationContext$UseCase[] values()
 ##
+public interface net.corda.core.serialization.SerializationCustomSerializer
+  public abstract Object fromProxy(Object)
+  public abstract Object toProxy(Object)
+##
 public final class net.corda.core.serialization.SerializationDefaults extends java.lang.Object
   @org.jetbrains.annotations.NotNull public final net.corda.core.serialization.SerializationContext getCHECKPOINT_CONTEXT()
   @org.jetbrains.annotations.NotNull public final net.corda.core.serialization.SerializationContext getP2P_CONTEXT()

client/rpc/src/main/kotlin/net/corda/client/rpc/internal/KryoClientSerializationScheme.kt

@@ -41,7 +41,7 @@ class KryoClientSerializationScheme : AbstractKryoSerializationScheme() {
             return SerializationEnvironmentImpl(
                     SerializationFactoryImpl().apply {
                         registerScheme(KryoClientSerializationScheme())
-                        registerScheme(AMQPClientSerializationScheme())
+                        registerScheme(AMQPClientSerializationScheme(emptyList()))
                     },
                     KRYO_P2P_CONTEXT,
                     rpcClientContext = KRYO_RPC_CLIENT_CONTEXT)

core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt

@@ -3,6 +3,7 @@ package net.corda.core.cordapp
 import net.corda.core.DoNotImplement
 import net.corda.core.flows.FlowLogic
 import net.corda.core.schemas.MappedSchema
+import net.corda.core.serialization.SerializationCustomSerializer
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.core.serialization.SerializeAsToken
 import java.net.URL
@@ -22,6 +23,7 @@ import java.net.URL
  * @property schedulableFlows List of flows startable by the scheduler
  * @property services List of RPC services
  * @property serializationWhitelists List of Corda plugin registries
+ * @property serializationCustomSerializers List of serializers
  * @property customSchemas List of custom schemas
  * @property jarPath The path to the JAR for this CorDapp
  */
@@ -35,6 +37,7 @@ interface Cordapp {
     val schedulableFlows: List<Class<out FlowLogic<*>>>
     val services: List<Class<out SerializeAsToken>>
     val serializationWhitelists: List<SerializationWhitelist>
+    val serializationCustomSerializers: List<SerializationCustomSerializer<*, *>>
     val customSchemas: Set<MappedSchema>
     val jarPath: URL
     val cordappClasses: List<String>

core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt

@@ -3,6 +3,7 @@ package net.corda.core.internal.cordapp
 import net.corda.core.cordapp.Cordapp
 import net.corda.core.flows.FlowLogic
 import net.corda.core.schemas.MappedSchema
+import net.corda.core.serialization.SerializationCustomSerializer
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.core.serialization.SerializeAsToken
 import java.io.File
@@ -16,6 +17,7 @@ data class CordappImpl(
         override val schedulableFlows: List<Class<out FlowLogic<*>>>,
         override val services: List<Class<out SerializeAsToken>>,
         override val serializationWhitelists: List<SerializationWhitelist>,
+        override val serializationCustomSerializers: List<SerializationCustomSerializer<*, *>>,
         override val customSchemas: Set<MappedSchema>,
         override val jarPath: URL) : Cordapp {
     override val name: String = File(jarPath.toURI()).name.removeSuffix(".jar")

core/src/main/kotlin/net/corda/core/serialization/SerializationCustomSerializer.kt

@@ -0,0 +1,24 @@
+package net.corda.core.serialization
+
+/**
+ * Allows CorDapps to provide custom serializers for third party libraries where those libraries cannot
+ * be recompiled with the -parameters flag rendering their classes natively serializable by Corda. In this case
+ * a proxy serializer can be written that extends this type whose purpose is to move between those an
+ * unserializable types and an intermediate representation.
+ *
+ * NOTE: The proxy object should be specified as a seperate class. However, this can be defined within the
+ * scope of the custom serializer.
+ */
+interface SerializationCustomSerializer<OBJ, PROXY> {
+    /**
+     * Should facilitate the conversion of the third party object into the serializable
+     * local class specified by [PROXY]
+     */
+    fun toProxy(obj: OBJ) : PROXY
+
+    /**
+     * Should facilitate the conversion of the proxy object into a new instance of the
+     * unserializable type
+     */
+    fun fromProxy(proxy: PROXY) : OBJ
+}

docs/source/cordapp-custom-serializers.rst

@@ -0,0 +1,73 @@
+Pluggable Serializers for CorDapps
+==================================
+
+.. contents::
+
+To be serializable by Corda Java classes must be compiled with the -parameters switch to enable matching of its properties
+to constructor parameters. This is important because Corda's internal AMQP serialization scheme will only construct
+objects using their constructors. However, when recompilation isn't possible, or classes are built in such a way that
+they cannot be easily modified for simple serialization, CorDapps can provide custom proxy serializers that Corda
+can use to move from types it cannot serialize to an interim representation that it can with the transformation to and
+from this proxy object being handled by the supplied serializer.
+
+Serializer Location
+-------------------
+Custom serializer classes should follow the rules for including classes found in :doc:`cordapp-build-systems`
+
+Writing a Custom Serializer
+---------------------------
+Serializers must
+ * Inherit from net.corda.core.serialization.SerializationCustomSerializer
+ * Provide a proxy class to transform the object to and from
+ * Implement the ``toProxy`` and ``fromProxy`` methods
+
+Serializers inheriting from SerializationCustomSerializer have to implement two methods and two types.
+
+Example
+-------
+Consider this example class
+
+
+.. sourcecode:: java
+
+    public final class Example {
+        private final Int a
+        private final Int b
+
+        private Example(Int a, Int b) {
+            this.a = a;
+            this.b = b;
+        }
+
+        public static Example of (int[] a) { return Example(a[0], a[1]); }
+
+        public int getA() { return a; }
+        public int getB() { return b; }
+    }
+
+Without a custom serializer we cannot serialize this class as there is no public constructor that facilitates the
+initialisation of all of its properties.
+
+To be serializable by Corda this would require a custom serializer as follows:
+
+.. sourcecode:: kotlin
+
+    class ExampleSerializer : SerializationCustomSerializer<Example, ExampleSerializer.Proxy> {
+        data class Proxy(val a: Int, val b: Int)
+
+        override fun toProxy(obj: Example) = Proxy(obj.a, obj.b)
+
+        override fun fromProxy(proxy: Proxy) : Example {
+            val constructorArg = IntArray(2);
+            constructorArg[0] = proxy.a
+            constructorArg[1] = proxy.b
+            return Example.create(constructorArg)
+        }
+    }
+
+Whitelisting
+------------
+By writing a custom serializer for a class it has the effect of adding that class to the whitelist, meaning such
+classes don't need explicitly adding to the CorDapp's whitelist.
+
+

docs/source/release-notes.rst

@@ -12,6 +12,20 @@ Unreleased
   That is the ability to alter an enum constant and, as long as certain rules are followed and the correct
   annotations applied, have older and newer instances of that enumeration be understood.
 
+* **AMQP Enabled**
+
+AMQP Serialization is now enabled for both peer to peer communication and writing states to the vault. This change
+brings a stable format Corda can support internally throughout it's lifetime that meets the needs of Corda and our
+users.
+
+* **Custom Serializers**
+
+To allow interop with third party libraries that cannot be recompiled we add functionality that allows custom serializers
+to be written for those classes. If needed, a proxy object can be created as an interim step that allows Corda's internal
+serializers to operate on those types.
+
+A good example of this is the SIMM valuation demo which has a number of such serializers defined in the plugin/customserializers package
+
 Release 2.0
 ----------
 Following quickly on the heels of the release of Corda 1.0, Corda version 2.0 consolidates

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/AMQPSerializationScheme.kt

@@ -2,6 +2,7 @@
 
 package net.corda.nodeapi.internal.serialization.amqp
 
+import net.corda.core.cordapp.Cordapp
 import net.corda.core.serialization.*
 import net.corda.core.utilities.ByteSequence
 import net.corda.nodeapi.internal.serialization.DefaultWhitelist
@@ -24,7 +25,8 @@ fun SerializerFactory.addToWhitelist(vararg types: Class<*>) {
     }
 }
 
-abstract class AbstractAMQPSerializationScheme : SerializationScheme {
+abstract class AbstractAMQPSerializationScheme(val cordappLoader: List<Cordapp>) : SerializationScheme {
+
     companion object {
         private val serializationWhitelists: List<SerializationWhitelist> by lazy {
             ServiceLoader.load(SerializationWhitelist::class.java, this::class.java.classLoader).toList() + DefaultWhitelist
@@ -62,8 +64,15 @@ abstract class AbstractAMQPSerializationScheme : SerializationScheme {
             register(net.corda.nodeapi.internal.serialization.amqp.custom.EnumSetSerializer(this))
             register(net.corda.nodeapi.internal.serialization.amqp.custom.ContractAttachmentSerializer(this))
         }
-        for (whitelistProvider in serializationWhitelists)
+        for (whitelistProvider in serializationWhitelists) {
             factory.addToWhitelist(*whitelistProvider.whitelist.toTypedArray())
+        }
+
+        for (loader in cordappLoader) {
+            for (schema in loader.serializationCustomSerializers) {
+                factory.registerExternal(CorDappCustomSerializer(schema, factory))
+            }
+        }
     }
 
     private val serializerFactoriesForContexts = ConcurrentHashMap<Pair<ClassWhitelist, ClassLoader>, SerializerFactory>()
@@ -97,11 +106,11 @@ abstract class AbstractAMQPSerializationScheme : SerializationScheme {
         return SerializationOutput(serializerFactory).serialize(obj)
     }
 
-    protected fun canDeserializeVersion(byteSequence: ByteSequence): Boolean = AMQP_ENABLED && byteSequence == AmqpHeaderV1_0
+    protected fun canDeserializeVersion(byteSequence: ByteSequence): Boolean = byteSequence == AmqpHeaderV1_0
 }
 
 // TODO: This will eventually cover server RPC as well and move to node module, but for now this is not implemented
-class AMQPServerSerializationScheme : AbstractAMQPSerializationScheme() {
+class AMQPServerSerializationScheme(cordapps: List<Cordapp> = emptyList()) : AbstractAMQPSerializationScheme(cordapps) {
     override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory {
         throw UnsupportedOperationException()
     }
@@ -118,7 +127,7 @@ class AMQPServerSerializationScheme : AbstractAMQPSerializationScheme() {
 }
 
 // TODO: This will eventually cover client RPC as well and move to client module, but for now this is not implemented
-class AMQPClientSerializationScheme : AbstractAMQPSerializationScheme() {
+class AMQPClientSerializationScheme(cordapps: List<Cordapp> = emptyList()) : AbstractAMQPSerializationScheme(cordapps) {
     override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory {
         TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
     }

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/CorDappCustomSerializer.kt

@@ -0,0 +1,86 @@
+package net.corda.nodeapi.internal.serialization.amqp
+
+import net.corda.core.internal.uncheckedCast
+import net.corda.core.serialization.SerializationCustomSerializer
+import net.corda.nodeapi.internal.serialization.amqp.SerializerFactory.Companion.nameForType
+import org.apache.qpid.proton.amqp.Symbol
+import org.apache.qpid.proton.codec.Data
+import java.io.NotSerializableException
+import java.lang.reflect.Type
+import kotlin.reflect.jvm.javaType
+import kotlin.reflect.jvm.jvmErasure
+
+/**
+ * Index into the types list of the parent type of the serializer object, should be the
+ * type that this object proxies for
+ */
+const val CORDAPP_TYPE = 0
+
+/**
+ * Index into the types list of the parent type of the serializer object, should be the
+ * type of the proxy object that we're using to represent the object we're proxying for
+ */
+const val PROXY_TYPE = 1
+
+/**
+ * Wrapper class for user provided serializers
+ *
+ * Through the CorDapp JAR scanner we will have a list of custom serializer types that implement
+ * the toProxy and fromProxy methods. This class takes an instance of one of those objects and
+ * embeds it within a serialization context associated with a serializer factory by creating
+ * and instance of this class and registering that with a [SerializerFactory]
+ *
+ * Proxy serializers should transform an unserializable class into a representation that we can serialize
+ *
+ * @property serializer in instance of a user written serialization proxy, normally scanned and loaded
+ * automatically
+ * @property type the Java [Type] of the class which this serializes, inferred via reflection of the
+ * [serializer]'s super type
+ * @property proxyType the Java [Type] of the class into which instances of [type] are proxied for use by
+ * the underlying serialization engine
+ *
+ * @param factory a [SerializerFactory] belonging to the context this serializer is being instantiated
+ * for
+ */
+class CorDappCustomSerializer(
+        private val serializer: SerializationCustomSerializer<*, *>,
+        factory: SerializerFactory) : AMQPSerializer<Any>, SerializerFor {
+    override val revealSubclassesInSchema: Boolean get() = false
+    private val types = serializer::class.supertypes.filter { it.jvmErasure == SerializationCustomSerializer::class }
+            .flatMap { it.arguments }
+            .map { it.type!!.javaType }
+
+    init {
+        if (types.size != 2) {
+            throw NotSerializableException("Unable to determine serializer parent types")
+        }
+    }
+
+    override val type = types[CORDAPP_TYPE]
+    val proxyType = types[PROXY_TYPE]
+    override val typeDescriptor = Symbol.valueOf("$DESCRIPTOR_DOMAIN:${nameForType(type)}")
+    val descriptor: Descriptor = Descriptor(typeDescriptor)
+    private val proxySerializer: ObjectSerializer by lazy { ObjectSerializer(proxyType, factory) }
+
+    override fun writeClassInfo(output: SerializationOutput) {}
+
+    override fun writeObject(obj: Any, data: Data, type: Type, output: SerializationOutput) {
+        val proxy = uncheckedCast<SerializationCustomSerializer<*, *>,
+                SerializationCustomSerializer<Any?, Any?>>(serializer).toProxy(obj)
+
+        data.withDescribed(descriptor) {
+            data.withList {
+                for (property in proxySerializer.propertySerializers) {
+                    property.writeProperty(proxy, this, output)
+                }
+            }
+        }
+    }
+
+    override fun readObject(obj: Any, schemas: SerializationSchemas, input: DeserializationInput) =
+            uncheckedCast<SerializationCustomSerializer<*, *>, SerializationCustomSerializer<Any?, Any?>>(
+                    serializer).fromProxy(uncheckedCast(proxySerializer.readObject(obj, schemas, input)))!!
+
+    override fun isSerializerFor(clazz: Class<*>) = clazz == type
+}
+

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/CustomSerializer.kt

@@ -6,22 +6,27 @@ import org.apache.qpid.proton.amqp.Symbol
 import org.apache.qpid.proton.codec.Data
 import java.lang.reflect.Type
 
+interface SerializerFor {
+    /**
+     * This method should return true if the custom serializer can serialize an instance of the class passed as the
+     * parameter.
+     */
+    fun isSerializerFor(clazz: Class<*>): Boolean
+
+    val revealSubclassesInSchema: Boolean
+}
+
 /**
  * Base class for serializers of core platform types that do not conform to the usual serialization rules and thus
  * cannot be automatically serialized.
  */
-abstract class CustomSerializer<T : Any> : AMQPSerializer<T> {
+abstract class CustomSerializer<T : Any> : AMQPSerializer<T>, SerializerFor {
     /**
      * This is a collection of custom serializers that this custom serializer depends on.  e.g. for proxy objects
      * that refer to other custom types etc.
      */
     open val additionalSerializers: Iterable<CustomSerializer<out Any>> = emptyList()
 
-    /**
-     * This method should return true if the custom serializer can serialize an instance of the class passed as the
-     * parameter.
-     */
-    abstract fun isSerializerFor(clazz: Class<*>): Boolean
 
     protected abstract val descriptor: Descriptor
     /**
@@ -33,7 +38,7 @@ abstract class CustomSerializer<T : Any> : AMQPSerializer<T> {
     /**
      * Whether subclasses using this serializer via inheritance should have a mapping in the schema.
      */
-    open val revealSubclassesInSchema: Boolean = false
+    override val revealSubclassesInSchema: Boolean get() = false
 
     override fun writeObject(obj: Any, data: Data, type: Type, output: SerializationOutput) {
         data.withDescribed(descriptor) {

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/SerializationHelper.kt

@@ -82,11 +82,13 @@ private fun <T : Any> propertiesForSerializationFromConstructor(kotlinConstructo
     for (param in kotlinConstructor.parameters) {
         val name = param.name ?: throw NotSerializableException("Constructor parameter of $clazz has no name.")
         val matchingProperty = properties[name] ?:
-                throw NotSerializableException("No property matching constructor parameter named '$name' of '$clazz'." +
+                throw NotSerializableException("No property matching constructor parameter named '$name' of '$clazz'. " +
-                        " If using Java, check that you have the -parameters option specified in the Java compiler.")
+                        "If using Java, check that you have the -parameters option specified in the Java compiler. " +
+                        "Alternately, provide a proxy serializer (SerializationCustomSerializer) if recompiling isn't an option")
         // Check that the method has a getter in java.
-        val getter = matchingProperty.readMethod ?: throw NotSerializableException("Property has no getter method for $name of $clazz." +
+        val getter = matchingProperty.readMethod ?: throw NotSerializableException("Property has no getter method for $name of $clazz. " +
-                " If using Java and the parameter name looks anonymous, check that you have the -parameters option specified in the Java compiler.")
+                "If using Java and the parameter name looks anonymous, check that you have the -parameters option specified in the Java compiler." +
+                "Alternately, provide a proxy serializer (SerializationCustomSerializer) if recompiling isn't an option")
         val returnType = resolveTypeVariables(getter.genericReturnType, type)
         if (constructorParamTakesReturnTypeOfGetter(returnType, getter.genericReturnType, param)) {
             rc += PropertySerializer.make(name, getter, returnType, factory)

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/SerializerFactory.kt

@@ -36,7 +36,7 @@ data class FactorySchemaAndDescriptor(val schemas: SerializationSchemas, val typ
 open class SerializerFactory(val whitelist: ClassWhitelist, cl: ClassLoader) {
     private val serializersByType = ConcurrentHashMap<Type, AMQPSerializer<Any>>()
     private val serializersByDescriptor = ConcurrentHashMap<Any, AMQPSerializer<Any>>()
-    private val customSerializers = CopyOnWriteArrayList<CustomSerializer<out Any>>()
+    private val customSerializers = CopyOnWriteArrayList<SerializerFor>()
     val transformsCache = ConcurrentHashMap<String, EnumMap<TransformTypes, MutableList<Transform>>>()
 
     open val classCarpenter = ClassCarpenter(cl, whitelist)
@@ -196,9 +196,16 @@ open class SerializerFactory(val whitelist: ClassWhitelist, cl: ClassLoader) {
         }
     }
 
+    fun registerExternal(customSerializer: CorDappCustomSerializer) {
+        if (!serializersByDescriptor.containsKey(customSerializer.typeDescriptor)) {
+            customSerializers += customSerializer
+            serializersByDescriptor[customSerializer.typeDescriptor] = customSerializer
+        }
+    }
+
     /**
-     * Iterate over an AMQP schema, for each type ascertain weather it's on ClassPath of [classloader] amd
+     * Iterate over an AMQP schema, for each type ascertain whether it's on ClassPath of [classloader] and,
-     * if not use the [ClassCarpenter] to generate a class to use in it's place
+     * if not, use the [ClassCarpenter] to generate a class to use in it's place.
      */
     private fun processSchema(schemaAndDescriptor: FactorySchemaAndDescriptor, sentinel: Boolean = false) {
         val metaSchema = CarpenterMetaSchema.newInstance()
@@ -267,11 +274,13 @@ open class SerializerFactory(val whitelist: ClassWhitelist, cl: ClassLoader) {
         for (customSerializer in customSerializers) {
             if (customSerializer.isSerializerFor(clazz)) {
                 val declaredSuperClass = declaredType.asClass()?.superclass
-                if (declaredSuperClass == null || !customSerializer.isSerializerFor(declaredSuperClass) || !customSerializer.revealSubclassesInSchema) {
+                return if (declaredSuperClass == null
-                    return customSerializer
+                        || !customSerializer.isSerializerFor(declaredSuperClass)
+                        || !customSerializer.revealSubclassesInSchema) {
+                    customSerializer as? AMQPSerializer<Any>
                 } else {
                     // Make a subclass serializer for the subclass and return that...
-                    return CustomSerializer.SubClass(clazz, uncheckedCast(customSerializer))
+                    CustomSerializer.SubClass(clazz, uncheckedCast(customSerializer))
                 }
             }
         }

node-api/src/test/kotlin/net/corda/nodeapi/internal/serialization/amqp/CorDappSerializerTests.kt

@@ -0,0 +1,148 @@
+package net.corda.nodeapi.internal.serialization.amqp
+
+import org.junit.Test
+import net.corda.core.serialization.ClassWhitelist
+import net.corda.core.serialization.SerializationCustomSerializer
+import org.assertj.core.api.Assertions
+import java.io.NotSerializableException
+import kotlin.test.assertEquals
+
+class CorDappSerializerTests {
+    data class NeedsProxy (val a: String)
+
+    class NeedsProxyProxySerializer : SerializationCustomSerializer<NeedsProxy, NeedsProxyProxySerializer.Proxy> {
+        data class Proxy(val proxy_a_: String)
+
+        override fun fromProxy(proxy: Proxy) = NeedsProxy(proxy.proxy_a_)
+        override fun toProxy(obj: NeedsProxy) = Proxy(obj.a)
+    }
+
+    // Standard proxy serializer used internally, here for comparison purposes
+    class InternalProxySerializer(factory: SerializerFactory) :
+            CustomSerializer.Proxy<NeedsProxy, InternalProxySerializer.Proxy> (
+                    NeedsProxy::class.java,
+                    InternalProxySerializer.Proxy::class.java,
+                    factory) {
+        data class Proxy(val proxy_a_: String)
+
+        override fun toProxy(obj: NeedsProxy): Proxy {
+            return Proxy(obj.a)
+        }
+
+        override fun fromProxy(proxy: Proxy): NeedsProxy {
+            return NeedsProxy(proxy.proxy_a_)
+        }
+    }
+
+    @Test
+    fun `type uses proxy`() {
+        val internalProxyFactory = testDefaultFactory()
+        val proxyFactory = testDefaultFactory()
+        val defaultFactory = testDefaultFactory()
+
+        val msg = "help"
+
+        proxyFactory.registerExternal (CorDappCustomSerializer(NeedsProxyProxySerializer(), proxyFactory))
+        internalProxyFactory.register (InternalProxySerializer(internalProxyFactory))
+
+        val needsProxy = NeedsProxy(msg)
+
+        val bAndSProxy = SerializationOutput(proxyFactory).serializeAndReturnSchema (needsProxy)
+        val bAndSInternal = SerializationOutput(internalProxyFactory).serializeAndReturnSchema (needsProxy)
+        val bAndSDefault = SerializationOutput(defaultFactory).serializeAndReturnSchema (needsProxy)
+
+        val objFromDefault = DeserializationInput(defaultFactory).deserializeAndReturnEnvelope(bAndSDefault.obj)
+        val objFromInternal = DeserializationInput(internalProxyFactory).deserializeAndReturnEnvelope(bAndSInternal.obj)
+        val objFromProxy = DeserializationInput(proxyFactory).deserializeAndReturnEnvelope(bAndSProxy.obj)
+
+        assertEquals(msg, objFromDefault.obj.a)
+        assertEquals(msg, objFromInternal.obj.a)
+        assertEquals(msg, objFromProxy.obj.a)
+    }
+
+    @Test
+    fun proxiedTypeIsNested() {
+        data class A (val a: Int, val b: NeedsProxy)
+
+        val factory = testDefaultFactory()
+        factory.registerExternal (CorDappCustomSerializer(NeedsProxyProxySerializer(), factory))
+
+        val tv1 = 100
+        val tv2 = "pants schmants"
+        val bAndS = SerializationOutput(factory).serializeAndReturnSchema (A(tv1, NeedsProxy(tv2)))
+
+        val objFromDefault = DeserializationInput(factory).deserializeAndReturnEnvelope(bAndS.obj)
+
+        assertEquals(tv1, objFromDefault.obj.a)
+        assertEquals(tv2, objFromDefault.obj.b.a)
+    }
+
+    @Test
+    fun testWithWhitelistNotAllowed() {
+        data class A (val a: Int, val b: NeedsProxy)
+
+        class WL : ClassWhitelist {
+            private val allowedClasses = emptySet<String>()
+
+            override fun hasListed(type: Class<*>): Boolean = type.name in allowedClasses
+        }
+
+        val factory = SerializerFactory(WL(), ClassLoader.getSystemClassLoader())
+        factory.registerExternal (CorDappCustomSerializer(NeedsProxyProxySerializer(), factory))
+
+        val tv1 = 100
+        val tv2 = "pants schmants"
+        Assertions.assertThatThrownBy {
+            SerializationOutput(factory).serialize(A(tv1, NeedsProxy(tv2)))
+        }.isInstanceOf(NotSerializableException::class.java)
+    }
+
+    @Test
+    fun testWithWhitelistAllowed() {
+        data class A (val a: Int, val b: NeedsProxy)
+
+        class WL : ClassWhitelist {
+            private val allowedClasses = hashSetOf(
+                    A::class.java.name,
+                    NeedsProxy::class.java.name)
+
+            override fun hasListed(type: Class<*>): Boolean = type.name in allowedClasses
+        }
+
+        val factory = SerializerFactory(WL(), ClassLoader.getSystemClassLoader())
+        factory.registerExternal (CorDappCustomSerializer(NeedsProxyProxySerializer(), factory))
+
+        val tv1 = 100
+        val tv2 = "pants schmants"
+        val obj = DeserializationInput(factory).deserialize(
+                SerializationOutput(factory).serialize(A(tv1, NeedsProxy(tv2))))
+
+        assertEquals(tv1, obj.a)
+        assertEquals(tv2, obj.b.a)
+    }
+
+    // The custom type not being whitelisted won't matter here because the act of adding a
+    // custom serializer bypasses the whitelist
+    @Test
+    fun testWithWhitelistAllowedOuterOnly() {
+        data class A (val a: Int, val b: NeedsProxy)
+
+        class WL : ClassWhitelist {
+            // explicitly don't add NeedsProxy
+            private val allowedClasses = hashSetOf(A::class.java.name)
+
+            override fun hasListed(type: Class<*>): Boolean = type.name in allowedClasses
+        }
+
+        val factory = SerializerFactory(WL(), ClassLoader.getSystemClassLoader())
+        factory.registerExternal (CorDappCustomSerializer(NeedsProxyProxySerializer(), factory))
+
+        val tv1 = 100
+        val tv2 = "pants schmants"
+        val obj = DeserializationInput(factory).deserialize(
+                SerializationOutput(factory).serialize(A(tv1, NeedsProxy(tv2))))
+
+        assertEquals(tv1, obj.a)
+        assertEquals(tv2, obj.b.a)
+    }
+}

node-api/src/test/kotlin/net/corda/nodeapi/internal/serialization/amqp/OverridePKSerializerTest.kt

@@ -25,7 +25,7 @@ class OverridePKSerializerTest {
             get() = TODO("not implemented") //To change initializer of created properties use File | Settings | File Templates.
     }
 
-    class AMQPTestSerializationScheme : AbstractAMQPSerializationScheme() {
+    class AMQPTestSerializationScheme : AbstractAMQPSerializationScheme(emptyList()) {
         override fun rpcServerSerializerFactory(context: SerializationContext): SerializerFactory {
             TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
         }

node-api/src/test/kotlin/net/corda/nodeapi/internal/serialization/amqp/SerializationOutputTests.kt

@@ -593,7 +593,7 @@ class SerializationOutputTests {
     fun `test transaction state`() {
         val state = TransactionState(FooState(), FOO_PROGRAM_ID, MEGA_CORP)
 
-        val scheme = AMQPServerSerializationScheme()
+        val scheme = AMQPServerSerializationScheme(emptyList())
         val func = scheme::class.superclasses.single { it.simpleName == "AbstractAMQPSerializationScheme" }
                 .java.getDeclaredMethod("registerCustomSerializers", SerializerFactory::class.java)
         func.isAccessible = true

node/src/main/kotlin/net/corda/node/internal/Node.kt

@@ -305,11 +305,11 @@ open class Node(configuration: NodeConfiguration,
         nodeSerializationEnv = SerializationEnvironmentImpl(
                 SerializationFactoryImpl().apply {
                     registerScheme(KryoServerSerializationScheme())
-                    registerScheme(AMQPServerSerializationScheme())
+                    registerScheme(AMQPServerSerializationScheme(cordappLoader.cordapps))
                 },
-                KRYO_P2P_CONTEXT.withClassLoader(classloader),
+                p2pContext = AMQP_P2P_CONTEXT.withClassLoader(classloader),
                 rpcServerContext = KRYO_RPC_SERVER_CONTEXT.withClassLoader(classloader),
-                storageContext = KRYO_STORAGE_CONTEXT.withClassLoader(classloader),
+                storageContext = AMQP_STORAGE_CONTEXT.withClassLoader(classloader),
                 checkpointContext = KRYO_CHECKPOINT_CONTEXT.withClassLoader(classloader))
     }
 

node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt

@@ -10,6 +10,7 @@ import net.corda.core.internal.*
 import net.corda.core.internal.cordapp.CordappImpl
 import net.corda.core.node.services.CordaService
 import net.corda.core.schemas.MappedSchema
+import net.corda.core.serialization.SerializationCustomSerializer
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.core.serialization.SerializeAsToken
 import net.corda.core.utilities.contextLogger
@@ -175,15 +176,16 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
         /** A Cordapp representing the core package which is not scanned automatically. */
         @VisibleForTesting
         internal val coreCordapp = CordappImpl(
-                listOf(),
+                contractClassNames = listOf(),
-                listOf(),
+                initiatedFlows = listOf(),
-                coreRPCFlows,
+                rpcFlows = coreRPCFlows,
-                listOf(),
+                serviceFlows = listOf(),
-                listOf(),
+                schedulableFlows = listOf(),
-                listOf(),
+                services = listOf(),
-                listOf(),
+                serializationWhitelists = listOf(),
-                setOf(),
+                serializationCustomSerializers = listOf(),
-                ContractUpgradeFlow.javaClass.protectionDomain.codeSource.location // Core JAR location
+                customSchemas = setOf(),
+                jarPath = ContractUpgradeFlow.javaClass.protectionDomain.codeSource.location // Core JAR location
         )
     }
 
@@ -197,6 +199,7 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
                     findSchedulableFlows(scanResult),
                     findServices(scanResult),
                     findPlugins(it),
+                    findSerializers(scanResult),
                     findCustomSchemas(scanResult),
                     it.url)
         }
@@ -247,6 +250,10 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
         } + DefaultWhitelist // Always add the DefaultWhitelist to the whitelist for an app.
     }
 
+    private fun findSerializers(scanResult: RestrictedScanResult): List<SerializationCustomSerializer<*, *>> {
+        return scanResult.getClassesImplementing(SerializationCustomSerializer::class)
+    }
+
     private fun findCustomSchemas(scanResult: RestrictedScanResult): Set<MappedSchema> {
         return scanResult.getClassesWithSuperclass(MappedSchema::class).toSet()
     }
@@ -303,6 +310,14 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
                     .map { it.kotlin.objectOrNewInstance() }
         }
 
+        fun <T : Any> getClassesImplementing(type: KClass<T>): List<T> {
+            return scanResult.getNamesOfClassesImplementing(type.java)
+                    .filter { it.startsWith(qualifiedNamePrefix) }
+                    .mapNotNull { loadClass(it, type) }
+                    .filterNot { Modifier.isAbstract(it.modifiers) }
+                    .map { it.kotlin.objectOrNewInstance() }
+        }
+
         fun <T : Any> getClassesWithAnnotation(type: KClass<T>, annotation: KClass<out Annotation>): List<Class<out T>> {
             return scanResult.getNamesOfClassesWithAnnotation(annotation.java)
                     .filter { it.startsWith(qualifiedNamePrefix) }

node/src/main/kotlin/net/corda/node/services/transactions/RaftUniquenessProvider.kt

@@ -239,4 +239,4 @@ private fun readMap(buffer: BufferInput<out BufferInput<*>>, serializer: Seriali
             put(serializer.readObject(buffer), serializer.readObject(buffer))
         }
     }
-}
+}

samples/simm-valuation-demo/src/integration-test/kotlin/net/corda/vega/SimmValuationTest.kt

@@ -27,7 +27,7 @@ class SimmValuationTest {
 
     @Test
     fun `runs SIMM valuation demo`() {
-        driver(isDebug = true, extraCordappPackagesToScan = listOf("net.corda.vega.contracts")) {
+        driver(isDebug = true, extraCordappPackagesToScan = listOf("net.corda.vega.contracts", "net.corda.vega.plugin.customserializers")) {
             val nodeAFuture = startNode(providedName = nodeALegalName)
             val nodeBFuture = startNode(providedName = nodeBLegalName)
             val (nodeA, nodeB) = listOf(nodeAFuture, nodeBFuture).map { it.getOrThrow() }

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/contracts/PortfolioState.kt

@@ -19,7 +19,7 @@ val PORTFOLIO_SWAP_PROGRAM_ID = "net.corda.vega.contracts.PortfolioSwap"
  * given point in time. This state can be consumed to create a new state with a mutated valuation or portfolio.
  */
 data class PortfolioState(val portfolio: List<StateRef>,
-                          private val _parties: Pair<AbstractParty, AbstractParty>,
+                          val _parties: Pair<AbstractParty, AbstractParty>,
                           val valuationDate: LocalDate,
                           val valuation: PortfolioValuation? = null,
                           override val linearId: UniqueIdentifier = UniqueIdentifier())

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/SimmPluginRegistry.kt

@@ -10,6 +10,7 @@ import com.opengamma.strata.market.curve.CurveName
 import com.opengamma.strata.market.param.CurrencyParameterSensitivities
 import com.opengamma.strata.market.param.CurrencyParameterSensitivity
 import com.opengamma.strata.market.param.TenorDateParameterMetadata
+import com.opengamma.strata.market.param.ParameterMetadata
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.vega.analytics.CordaMarketData
 import net.corda.vega.analytics.InitialMarginTriple
@@ -34,6 +35,7 @@ class SimmPluginRegistry : SerializationWhitelist {
             DoubleArray::class.java,
             CurveName::class.java,
             TenorDateParameterMetadata::class.java,
-            Tenor::class.java
+            Tenor::class.java,
+            ParameterMetadata::class.java
     )
 }

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/CurrencyParameterSensitivitiesSerializer.kt

@@ -0,0 +1,14 @@
+package net.corda.vega.plugin.customserializers
+
+import com.opengamma.strata.market.param.CurrencyParameterSensitivities
+import com.opengamma.strata.market.param.CurrencyParameterSensitivity
+import net.corda.core.serialization.SerializationCustomSerializer
+
+@Suppress("UNUSED")
+class CurrencyParameterSensitivitiesSerializer :
+        SerializationCustomSerializer<CurrencyParameterSensitivities, CurrencyParameterSensitivitiesSerializer.Proxy> {
+    data class Proxy(val sensitivities: List<CurrencyParameterSensitivity>)
+
+    override fun fromProxy(proxy: Proxy) = CurrencyParameterSensitivities.of(proxy.sensitivities)
+    override fun toProxy(obj: CurrencyParameterSensitivities) = Proxy(obj.sensitivities)
+}

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/CurrencyParameterSensitivitySerialiser.kt

@@ -0,0 +1,26 @@
+package net.corda.vega.plugin.customserializers
+
+import com.opengamma.strata.market.param.CurrencyParameterSensitivity
+import com.opengamma.strata.market.param.ParameterMetadata
+import com.opengamma.strata.data.MarketDataName
+import com.opengamma.strata.collect.array.DoubleArray
+import com.opengamma.strata.basics.currency.Currency
+import net.corda.core.serialization.SerializationCustomSerializer
+
+@Suppress("UNUSED")
+class CurrencyParameterSensitivitySerializer :
+        SerializationCustomSerializer<CurrencyParameterSensitivity, CurrencyParameterSensitivitySerializer.Proxy> {
+    data class Proxy(val currency: Currency, val marketDataName: MarketDataName<*>,
+                     val parameterMetadata: List<ParameterMetadata>,
+                     val sensitivity: DoubleArray)
+
+    override fun fromProxy(proxy: CurrencyParameterSensitivitySerializer.Proxy) =
+            CurrencyParameterSensitivity.of(
+                    proxy.marketDataName,
+                    proxy.parameterMetadata,
+                    proxy.currency,
+                    proxy.sensitivity)
+
+    override fun toProxy(obj: CurrencyParameterSensitivity) = Proxy((obj as CurrencyParameterSensitivity).currency,
+            obj.marketDataName, obj.parameterMetadata, obj.sensitivity)
+}

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/CurrencySerializer.kt

@@ -0,0 +1,12 @@
+package net.corda.vega.plugin.customserializers
+
+import com.opengamma.strata.basics.currency.Currency
+import net.corda.core.serialization.SerializationCustomSerializer
+
+@Suppress("UNUSED")
+class CurrencySerializer : SerializationCustomSerializer<Currency, CurrencySerializer.Proxy> {
+    data class Proxy(val currency: String)
+
+    override fun fromProxy(proxy: Proxy) = Currency.parse(proxy.currency)
+    override fun toProxy(obj: Currency) = Proxy(obj.toString())
+}

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/DoubleArraySerializer.kt

@@ -0,0 +1,12 @@
+package net.corda.vega.plugin.customserializers
+
+import net.corda.core.serialization.SerializationCustomSerializer
+import com.opengamma.strata.collect.array.DoubleArray
+
+@Suppress("UNUSED")
+class DoubleArraySerializer : SerializationCustomSerializer<DoubleArray, DoubleArraySerializer.Proxy> {
+    data class Proxy(val amount: kotlin.DoubleArray)
+
+    override fun fromProxy(proxy: Proxy) = DoubleArray.copyOf(proxy.amount)
+    override fun toProxy(obj: DoubleArray) = Proxy(obj.toArray())
+}

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/MultiCurrencyAmountSerializer.kt

@@ -0,0 +1,17 @@
+package net.corda.vega.plugin.customserializers
+
+import com.opengamma.strata.basics.currency.MultiCurrencyAmount
+import com.opengamma.strata.basics.currency.Currency
+import net.corda.core.serialization.*
+
+@Suppress("UNUSED")
+class MultiCurrencyAmountSerializer :
+        SerializationCustomSerializer<MultiCurrencyAmount, MultiCurrencyAmountSerializer.Proxy> {
+    data class Proxy(val curencies : Map<Currency, Double>)
+
+    override fun toProxy(obj: MultiCurrencyAmount) = Proxy(obj.toMap())
+    override fun fromProxy(proxy: Proxy) = MultiCurrencyAmount.of(proxy.curencies)
+}
+
+
+

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/TenorDateParameterMetadataSerializer.kt

@@ -0,0 +1,15 @@
+package net.corda.vega.plugin.customserializers
+
+import com.opengamma.strata.basics.date.Tenor
+import com.opengamma.strata.market.param.TenorDateParameterMetadata
+import net.corda.core.serialization.*
+import java.time.LocalDate
+
+@Suppress("UNUSED")
+class TenorDateParameterMetadataSerializer :
+        SerializationCustomSerializer<TenorDateParameterMetadata, TenorDateParameterMetadataSerializer.Proxy> {
+    data class Proxy(val tenor: Tenor, val date: LocalDate, val identifier: Tenor, val label: String)
+
+    override fun toProxy(obj: TenorDateParameterMetadata) = Proxy(obj.tenor, obj.date, obj.identifier, obj.label)
+    override fun fromProxy(proxy: Proxy) = TenorDateParameterMetadata.of(proxy.date, proxy.tenor, proxy.label)
+}

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/plugin/customserializers/TenorSerializer.kt

@@ -0,0 +1,13 @@
+package net.corda.vega.plugin.customserializers
+
+import com.opengamma.strata.basics.date.Tenor
+import net.corda.core.serialization.*
+import java.time.Period
+
+@Suppress("UNUSED")
+class TenorSerializer : SerializationCustomSerializer<Tenor, TenorSerializer.Proxy> {
+    data class Proxy(val years: Int, val months: Int, val days: Int, val name: String)
+
+    override fun toProxy(obj: Tenor) = Proxy(obj.period.years, obj.period.months, obj.period.days, obj.toString())
+    override fun fromProxy(proxy: Proxy) = Tenor.of (Period.of(proxy.years, proxy.months, proxy.days))
+}

testing/node-driver/src/main/kotlin/net/corda/testing/node/MockNode.kt

@@ -47,6 +47,8 @@ import net.corda.testing.testNodeConfiguration
 import org.apache.activemq.artemis.utils.ReusableLatch
 import org.apache.sshd.common.util.security.SecurityUtils
 import rx.internal.schedulers.CachedThreadScheduler
+import org.slf4j.Logger
+import java.io.Closeable
 import java.math.BigInteger
 import java.nio.file.Path
 import java.security.KeyPair

testing/test-utils/src/main/kotlin/net/corda/testing/SerializationTestHelpers.kt

@@ -62,7 +62,7 @@ fun <T> withTestSerialization(inheritable: Boolean = false, callable: (Serializa
 
 /**
  * For example your test class uses [SerializationEnvironmentRule] but you want to turn it off for one method.
- * Use sparingly, ideally a test class shouldn't mix serialization init mechanisms.
+ * Use sparingly, ideally a test class shouldn't mix serializers init mechanisms.
  */
 fun <T> withoutTestSerialization(callable: () -> T): T {
     val (property, env) = listOf(_contextSerializationEnv, _inheritableContextSerializationEnv).map { Pair(it, it.get()) }.single { it.second != null }
@@ -99,13 +99,13 @@ private fun createTestSerializationEnv(label: String) = object : SerializationEn
         SerializationFactoryImpl().apply {
             registerScheme(KryoClientSerializationScheme())
             registerScheme(KryoServerSerializationScheme())
-            registerScheme(AMQPClientSerializationScheme())
+            registerScheme(AMQPClientSerializationScheme(emptyList()))
-            registerScheme(AMQPServerSerializationScheme())
+            registerScheme(AMQPServerSerializationScheme(emptyList()))
         },
-        if (isAmqpEnabled()) AMQP_P2P_CONTEXT else KRYO_P2P_CONTEXT,
+        AMQP_P2P_CONTEXT,
         KRYO_RPC_SERVER_CONTEXT,
         KRYO_RPC_CLIENT_CONTEXT,
-        if (isAmqpEnabled()) AMQP_STORAGE_CONTEXT else KRYO_STORAGE_CONTEXT,
+        AMQP_STORAGE_CONTEXT,
         KRYO_CHECKPOINT_CONTEXT) {
     override fun toString() = "testSerializationEnv($label)"
 }

testing/test-utils/src/main/kotlin/net/corda/testing/node/MockCordappProvider.kt

@@ -14,7 +14,17 @@ class MockCordappProvider(cordappLoader: CordappLoader, attachmentStorage: Attac
     val cordappRegistry = mutableListOf<Pair<Cordapp, AttachmentId>>()
 
     fun addMockCordapp(contractClassName: ContractClassName, attachments: MockAttachmentStorage) {
-        val cordapp = CordappImpl(listOf(contractClassName), emptyList(), emptyList(), emptyList(), emptyList(), emptyList(), emptyList(), emptySet(), Paths.get(".").toUri().toURL())
+        val cordapp = CordappImpl(
+                contractClassNames = listOf(contractClassName),
+                initiatedFlows = emptyList(),
+                rpcFlows = emptyList(),
+                serviceFlows = emptyList(),
+                schedulableFlows = emptyList(),
+                services = emptyList(),
+                serializationWhitelists = emptyList(),
+                serializationCustomSerializers = emptyList(),
+                customSchemas = emptySet(),
+                jarPath = Paths.get(".").toUri().toURL())
         if (cordappRegistry.none { it.first.contractClassNames.contains(contractClassName) }) {
             cordappRegistry.add(Pair(cordapp, findOrImportAttachment(contractClassName.toByteArray(), attachments)))
         }

verifier/src/main/kotlin/net/corda/verifier/Verifier.kt

@@ -91,11 +91,7 @@ class Verifier {
                         registerScheme(KryoVerifierSerializationScheme)
                         registerScheme(AMQPVerifierSerializationScheme)
                     },
-                    /**
+                    AMQP_P2P_CONTEXT)
-                     * Even though default context is set to Kryo P2P, the encoding will be adjusted depending on the incoming
-                     * request received, see use of [context] in [main] method.
-                     */
-                    KRYO_P2P_CONTEXT)
         }
     }
 
@@ -108,7 +104,7 @@ class Verifier {
         override fun rpcServerKryoPool(context: SerializationContext) = throw UnsupportedOperationException()
     }
 
-    private object AMQPVerifierSerializationScheme : AbstractAMQPSerializationScheme() {
+    private object AMQPVerifierSerializationScheme : AbstractAMQPSerializationScheme(emptyList()) {
         override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
             return (byteSequence == AmqpHeaderV1_0 && (target == SerializationContext.UseCase.P2P))
         }
@@ -116,4 +112,4 @@ class Verifier {
         override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory = throw UnsupportedOperationException()
         override fun rpcServerSerializerFactory(context: SerializationContext): SerializerFactory = throw UnsupportedOperationException()
     }
-}
+}