CORDA-3680: Add CorDapp custom serialisers to Driver's in-process nodes. (#6098)

* Run serialisation tests with both in-process and out-of-process nodes.

* Add custom serialisers and whitelists to Driver's AMQPServerSerializationScheme.

client/rpc/src/main/kotlin/net/corda/client/rpc/internal/serialization/amqp/AMQPClientSerializationScheme.kt

@@ -8,7 +8,6 @@ import net.corda.core.serialization.SerializationCustomSerializer
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.core.serialization.internal.SerializationEnvironment
 import net.corda.core.serialization.internal._rpcClientSerializationEnv
-import net.corda.core.serialization.internal.nodeSerializationEnv
 import net.corda.serialization.internal.*
 import net.corda.serialization.internal.amqp.*
 import net.corda.serialization.internal.amqp.custom.RxNotificationSerializer
@@ -57,7 +56,7 @@ class AMQPClientSerializationScheme(
         }
     }
 
-    override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
+    override fun canDeserializeVersion(magic: CordaSerializationMagic, target: UseCase): Boolean {
         return magic == amqpMagic && (target == UseCase.RPCClient || target == UseCase.P2P)
     }
 

node/src/integration-test/kotlin/net/corda/node/ContractWithCustomSerializerTest.kt

@@ -18,13 +18,21 @@ import net.corda.testing.node.internal.cordappWithPackages
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.BeforeClass
 import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
 import kotlin.test.assertFailsWith
 
+@RunWith(Parameterized::class)
 @Suppress("FunctionName")
-class ContractWithCustomSerializerTest {
+class ContractWithCustomSerializerTest(private val runInProcess: Boolean) {
     companion object {
         const val CURRANTS = 5000L
 
+        @Parameters
+        @JvmStatic
+        fun modes(): List<Array<Boolean>> = listOf(Array(1) { true }, Array(1) { false })
+
         @BeforeClass
         @JvmStatic
         fun checkData() {
@@ -37,7 +45,7 @@ class ContractWithCustomSerializerTest {
         val user = User("u", "p", setOf(Permissions.all()))
         driver(DriverParameters(
             portAllocation = incrementalPortAllocation(),
-            startNodesInProcess = false,
+            startNodesInProcess = runInProcess,
             notarySpecs = listOf(NotarySpec(DUMMY_NOTARY_NAME, validating = true)),
             cordappsForAllNodes = listOf(
                 cordappWithPackages("net.corda.flows.serialization.custom").signed(),

node/src/integration-test/kotlin/net/corda/node/ContractWithMissingCustomSerializerTest.kt

@@ -26,10 +26,14 @@ import net.corda.testing.node.internal.cordappWithPackages
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.BeforeClass
 import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
 import kotlin.test.assertFailsWith
 
+@RunWith(Parameterized::class)
 @Suppress("FunctionName")
-class ContractWithMissingCustomSerializerTest {
+class ContractWithMissingCustomSerializerTest(private val runInProcess: Boolean) {
     companion object {
         const val BOBBINS = 5000L
 
@@ -37,15 +41,19 @@ class ContractWithMissingCustomSerializerTest {
         val flowCorDapp = cordappWithPackages("net.corda.flows.serialization.missing").signed()
         val contractCorDapp = cordappWithPackages("net.corda.contracts.serialization.missing").signed()
 
-        fun driverParameters(cordapps: List<TestCordapp>): DriverParameters {
+        fun driverParameters(cordapps: List<TestCordapp>, runInProcess: Boolean): DriverParameters {
             return DriverParameters(
                 portAllocation = incrementalPortAllocation(),
-                startNodesInProcess = false,
+                startNodesInProcess = runInProcess,
                 notarySpecs = listOf(NotarySpec(DUMMY_NOTARY_NAME, validating = true)),
                 cordappsForAllNodes = cordapps
             )
         }
 
+        @Parameters
+        @JvmStatic
+        fun modes(): List<Array<Boolean>> = listOf(Array(1) { true }, Array(1) { false })
+
         @BeforeClass
         @JvmStatic
         fun checkData() {
@@ -62,7 +70,7 @@ class ContractWithMissingCustomSerializerTest {
         val flowId = flowCorDapp.jarFile.hash
         val fixupCorDapp = cordappWithFixups(listOf(setOf(contractId) to setOf(contractId, flowId))).signed()
 
-        driver(driverParameters(listOf(flowCorDapp, contractCorDapp, fixupCorDapp))) {
+        driver(driverParameters(listOf(flowCorDapp, contractCorDapp, fixupCorDapp), runInProcess)) {
             val alice = startNode(providedName = ALICE_NAME, rpcUsers = listOf(user)).getOrThrow()
             val ex = assertFailsWith<ContractRejection> {
                 CordaRPCClient(hostAndPort = alice.rpcAddress)
@@ -83,7 +91,7 @@ class ContractWithMissingCustomSerializerTest {
      */
     @Test(timeout=300_000)
 	fun `flow with missing custom serializer but without fixup`() {
-        driver(driverParameters(listOf(flowCorDapp, contractCorDapp))) {
+        driver(driverParameters(listOf(flowCorDapp, contractCorDapp), runInProcess)) {
             val alice = startNode(providedName = ALICE_NAME, rpcUsers = listOf(user)).getOrThrow()
             val ex = assertFailsWith<BrokenTransactionException> {
                 CordaRPCClient(hostAndPort = alice.rpcAddress)
@@ -104,7 +112,7 @@ class ContractWithMissingCustomSerializerTest {
      */
     @Test(timeout=300_000)
 	fun `transaction builder flow with missing custom serializer by rpc`() {
-        driver(driverParameters(listOf(flowCorDapp, contractCorDapp))) {
+        driver(driverParameters(listOf(flowCorDapp, contractCorDapp), runInProcess)) {
             val alice = startNode(providedName = ALICE_NAME, rpcUsers = listOf(user)).getOrThrow()
             val ex = assertFailsWith<CordaRuntimeException> {
                 CordaRPCClient(hostAndPort = alice.rpcAddress)

node/src/main/kotlin/net/corda/node/serialization/amqp/AMQPServerSerializationScheme.kt

@@ -18,10 +18,16 @@ class AMQPServerSerializationScheme(
         cordappSerializationWhitelists: Set<SerializationWhitelist>,
         serializerFactoriesForContexts: MutableMap<SerializationFactoryCacheKey, SerializerFactory>
 ) : AbstractAMQPSerializationScheme(cordappCustomSerializers, cordappSerializationWhitelists, serializerFactoriesForContexts) {
-    constructor(cordapps: List<Cordapp>) : this(cordapps.customSerializers, cordapps.serializationWhitelists, AccessOrderLinkedHashMap<SerializationFactoryCacheKey, SerializerFactory>(128).toSynchronised())
+    constructor(cordapps: List<Cordapp>) : this(cordapps.customSerializers, cordapps.serializationWhitelists)
-    constructor(cordapps: List<Cordapp>, serializerFactoriesForContexts: MutableMap<SerializationFactoryCacheKey, SerializerFactory>) : this(cordapps.customSerializers, cordapps.serializationWhitelists, serializerFactoriesForContexts)
+    constructor(cordapps: List<Cordapp>, serializerFactoriesForContexts: MutableMap<SerializationFactoryCacheKey, SerializerFactory>)
+        : this(cordapps.customSerializers, cordapps.serializationWhitelists, serializerFactoriesForContexts)
+    constructor(
+        cordappCustomSerializers: Set<SerializationCustomSerializer<*,*>>,
+        cordappSerializationWhitelists: Set<SerializationWhitelist>
+    ) : this(cordappCustomSerializers, cordappSerializationWhitelists, AccessOrderLinkedHashMap<SerializationFactoryCacheKey, SerializerFactory>(128).toSynchronised())
 
-    constructor() : this(emptySet(), emptySet(), AccessOrderLinkedHashMap<SerializationFactoryCacheKey, SerializerFactory>(128).toSynchronised() )
+    @Suppress("UNUSED")
+    constructor() : this(emptySet(), emptySet())
 
     override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory {
         throw UnsupportedOperationException()

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

@@ -21,16 +21,18 @@ fun createTestSerializationEnv(): SerializationEnvironment {
 }
 
 fun createTestSerializationEnv(classLoader: ClassLoader?): SerializationEnvironment {
-    val clientSerializationScheme = if (classLoader != null) {
+    val (clientSerializationScheme, serverSerializationScheme) = if (classLoader != null) {
         val customSerializers = createInstancesOfClassesImplementing(classLoader, SerializationCustomSerializer::class.java)
         val serializationWhitelists = ServiceLoader.load(SerializationWhitelist::class.java, classLoader).toSet()
-        AMQPClientSerializationScheme(customSerializers, serializationWhitelists)
+
+        Pair(AMQPClientSerializationScheme(customSerializers, serializationWhitelists),
+             AMQPServerSerializationScheme(customSerializers, serializationWhitelists))
     } else {
-        AMQPClientSerializationScheme(emptyList())
+        Pair(AMQPClientSerializationScheme(emptyList()), AMQPServerSerializationScheme(emptyList()))
     }
     val factory = SerializationFactoryImpl().apply {
         registerScheme(clientSerializationScheme)
-        registerScheme(AMQPServerSerializationScheme(emptyList()))
+        registerScheme(serverSerializationScheme)
     }
     return SerializationEnvironment.with(
             factory,