[CORDA-2431] - Small refactorings following-up on PR-4551 (#4564)

* Small refactorings following-up on PR-4551

* Adjust thread context class loader

* Address Shams' comments

client/rpc/src/main/kotlin/net/corda/client/rpc/CordaRPCClient.kt

@@ -3,11 +3,11 @@ package net.corda.client.rpc
 import com.github.benmanes.caffeine.cache.Caffeine
 import net.corda.client.rpc.internal.RPCClient
 import net.corda.client.rpc.internal.serialization.amqp.AMQPClientSerializationScheme
+import net.corda.core.internal.createInstancesOfClassesImplementing
 import net.corda.core.context.Actor
 import net.corda.core.context.Trace
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.internal.PLATFORM_VERSION
-import net.corda.core.internal.loadClassesImplementing
 import net.corda.core.messaging.ClientRpcSslOptions
 import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.serialization.SerializationCustomSerializer
@@ -22,7 +22,8 @@ import net.corda.serialization.internal.AMQP_RPC_CLIENT_CONTEXT
 import net.corda.serialization.internal.amqp.SerializationFactoryCacheKey
 import net.corda.serialization.internal.amqp.SerializerFactory
 import java.time.Duration
-import java.util.*
+import java.util.ServiceLoader
+import java.net.URLClassLoader
 
 /**
  * This class is essentially just a wrapper for an RPCConnection<CordaRPCOps> and can be treated identically.
@@ -244,11 +245,11 @@ open class CordaRPCClientConfiguration @JvmOverloads constructor(
  * @param configuration An optional configuration used to tweak client behaviour.
  * @param sslConfiguration An optional [ClientRpcSslOptions] used to enable secure communication with the server.
  * @param haAddressPool A list of [NetworkHostAndPort] representing the addresses of servers in HA mode.
- * The client will attempt to connect to a live server by trying each address in the list. If the servers are not in
- * HA mode, the client will round-robin from the beginning of the list and try all servers.
+ *  The client will attempt to connect to a live server by trying each address in the list. If the servers are not in
+ *  HA mode, the client will round-robin from the beginning of the list and try all servers.
  * @param classLoader a classloader, which will be used (if provided) to discover available [SerializationCustomSerializer]s and [SerializationWhitelist]s
- * The client will attempt to connect to a live server by trying each address in the list. If the servers are not in
- * HA mode, the client will round-robin from the beginning of the list and try all servers.
+ *  If the created RPC client is intended to use types with custom serializers / whitelists,
+ *  a classloader will need to be provided that contains the associated CorDapp jars.
  */
 class CordaRPCClient private constructor(
         private val hostAndPort: NetworkHostAndPort?,
@@ -257,11 +258,16 @@ class CordaRPCClient private constructor(
         private val sslConfiguration: ClientRpcSslOptions? = null,
         private val classLoader: ClassLoader? = null
 ) {
+
     @JvmOverloads
     constructor(hostAndPort: NetworkHostAndPort, configuration: CordaRPCClientConfiguration = CordaRPCClientConfiguration.DEFAULT) : this(
             hostAndPort = hostAndPort, haAddressPool = emptyList(), configuration = configuration
     )
 
+    constructor(hostAndPort: NetworkHostAndPort,
+                configuration: CordaRPCClientConfiguration = CordaRPCClientConfiguration.DEFAULT,
+                classLoader: ClassLoader): this(hostAndPort, configuration, null, classLoader = classLoader)
+
     constructor(
             hostAndPort: NetworkHostAndPort,
             sslConfiguration: ClientRpcSslOptions? = null,
@@ -292,13 +298,15 @@ class CordaRPCClient private constructor(
             effectiveSerializationEnv
         } catch (e: IllegalStateException) {
             try {
+                val cache = Caffeine.newBuilder().maximumSize(128).build<SerializationFactoryCacheKey, SerializerFactory>().asMap()
+
                 // If the client has provided a classloader, the associated classpath is checked for available custom serializers and serialization whitelists.
                 if (classLoader != null) {
-                    val customSerializers = loadClassesImplementing(classLoader, SerializationCustomSerializer::class.java)
+                    val customSerializers = createInstancesOfClassesImplementing(classLoader, SerializationCustomSerializer::class.java)
                     val serializationWhitelists = ServiceLoader.load(SerializationWhitelist::class.java, classLoader).toSet()
-                    AMQPClientSerializationScheme.initialiseSerialization(classLoader, customSerializers, serializationWhitelists, Caffeine.newBuilder().maximumSize(128).build<SerializationFactoryCacheKey, SerializerFactory>().asMap())
+                    AMQPClientSerializationScheme.initialiseSerialization(classLoader, customSerializers, serializationWhitelists, cache)
                 } else {
-                    AMQPClientSerializationScheme.initialiseSerialization(classLoader, serializerFactoriesForContexts =  Caffeine.newBuilder().maximumSize(128).build<SerializationFactoryCacheKey, SerializerFactory>().asMap())
+                    AMQPClientSerializationScheme.initialiseSerialization(classLoader, serializerFactoriesForContexts =  cache)
                 }
             } catch (e: IllegalStateException) {
                 // Race e.g. two of these constructed in parallel, ignore.

core-deterministic/testing/src/test/kotlin/net/corda/core/internal/ClassLoadingUtils.kt

@@ -3,6 +3,6 @@ package net.corda.core.internal
 /**
  * Stubbing out non-deterministic method.
  */
-fun <T: Any> loadClassesImplementing(classloader: ClassLoader, clazz: Class<T>): Set<T> {
+fun <T: Any> createInstancesOfClassesImplementing(classloader: ClassLoader, clazz: Class<T>): Set<T> {
     return emptySet()
 }

core/src/main/kotlin/net/corda/core/internal/ClassLoadingUtils.kt

@@ -2,7 +2,6 @@ package net.corda.core.internal
 
 import io.github.classgraph.ClassGraph
 import net.corda.core.StubOutForDJVM
-import kotlin.reflect.full.createInstance
 
 /**
  * Creates instances of all the classes in the classpath of the provided classloader, which implement the interface of the provided class.
@@ -17,15 +16,26 @@ import kotlin.reflect.full.createInstance
  * - either be a Kotlin object or have a constructor with no parameters (or only optional ones)
  */
 @StubOutForDJVM
-fun <T: Any> loadClassesImplementing(classloader: ClassLoader, clazz: Class<T>): Set<T> {
+fun <T: Any> createInstancesOfClassesImplementing(classloader: ClassLoader, clazz: Class<T>): Set<T> {
     return ClassGraph().addClassLoader(classloader)
             .enableClassInfo()
             .scan()
             .use {
                 it.getClassesImplementing(clazz.name)
                         .filterNot { it.isAbstract }
-                        .mapNotNull { classloader.loadClass(it.name).asSubclass(clazz) }
-                        .map { it.kotlin.objectInstance ?: it.kotlin.createInstance() }
+                        .map { classloader.loadClass(it.name).asSubclass(clazz) }
+                        .map { it.kotlin.objectOrNewInstance() }
                         .toSet()
             }
+}
+
+fun <T: Any?> executeWithThreadContextClassLoader(classloader: ClassLoader, fn: () -> T): T {
+    val threadClassLoader = Thread.currentThread().contextClassLoader
+    try {
+        Thread.currentThread().contextClassLoader = classloader
+        return fn()
+    } finally {
+        Thread.currentThread().contextClassLoader = threadClassLoader
+    }
+
 }

core/src/main/kotlin/net/corda/core/serialization/internal/AttachmentsClassLoader.kt

@@ -2,7 +2,7 @@ package net.corda.core.serialization.internal
 
 import net.corda.core.CordaException
 import net.corda.core.KeepForDJVM
-import net.corda.core.internal.loadClassesImplementing
+import net.corda.core.internal.createInstancesOfClassesImplementing
 import net.corda.core.contracts.Attachment
 import net.corda.core.contracts.ContractAttachment
 import net.corda.core.contracts.TransactionVerificationException.OverlappingAttachmentsException
@@ -11,7 +11,6 @@ import net.corda.core.crypto.sha256
 import net.corda.core.internal.*
 import net.corda.core.internal.cordapp.targetPlatformVersion
 import net.corda.core.serialization.CordaSerializable
-import net.corda.core.serialization.MissingAttachmentsException
 import net.corda.core.serialization.SerializationCustomSerializer
 import net.corda.core.serialization.SerializationFactory
 import net.corda.core.serialization.SerializationWhitelist
@@ -196,7 +195,7 @@ internal object AttachmentsClassLoaderBuilder {
         val serializationContext = cache.computeIfAbsent(attachmentIds) {
             // Create classloader and load serializers, whitelisted classes
             val transactionClassLoader = AttachmentsClassLoader(attachments)
-            val serializers = loadClassesImplementing(transactionClassLoader, SerializationCustomSerializer::class.java)
+            val serializers = createInstancesOfClassesImplementing(transactionClassLoader, SerializationCustomSerializer::class.java)
             val whitelistedClasses = ServiceLoader.load(SerializationWhitelist::class.java, transactionClassLoader)
                     .flatMap { it.whitelist }
                     .toList()

core/src/test/kotlin/net/corda/core/internal/ClassLoadingUtilsTest.kt

@@ -1,11 +1,16 @@
 package net.corda.core.internal
 
+import com.nhaarman.mockito_kotlin.mock
 import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.Test
 import java.lang.IllegalArgumentException
+import java.lang.RuntimeException
 
 class ClassLoadingUtilsTest {
 
+    private val temporaryClassLoader = mock<ClassLoader>()
+
     interface BaseInterface {}
 
     interface BaseInterface2 {}
@@ -18,7 +23,7 @@ class ClassLoadingUtilsTest {
 
     @Test
     fun predicateClassAreLoadedSuccessfully() {
-        val classes = loadClassesImplementing(BaseInterface::class.java.classLoader, BaseInterface::class.java)
+        val classes = createInstancesOfClassesImplementing(BaseInterface::class.java.classLoader, BaseInterface::class.java)
 
         val classNames = classes.map { it.javaClass.name }
 
@@ -28,7 +33,27 @@ class ClassLoadingUtilsTest {
 
     @Test(expected = IllegalArgumentException::class)
     fun throwsExceptionWhenClassDoesNotContainProperConstructors() {
-        val classes = loadClassesImplementing(BaseInterface::class.java.classLoader, BaseInterface2::class.java)
+        val classes = createInstancesOfClassesImplementing(BaseInterface::class.java.classLoader, BaseInterface2::class.java)
+    }
+
+    @Test
+    fun `thread context class loader is adjusted, during the function execution`() {
+        val result = executeWithThreadContextClassLoader(temporaryClassLoader) {
+            assertThat(Thread.currentThread().contextClassLoader).isEqualTo(temporaryClassLoader)
+            true
+        }
+
+        assertThat(result).isTrue()
+        assertThat(Thread.currentThread().contextClassLoader).isNotEqualTo(temporaryClassLoader)
+    }
+
+    @Test
+    fun `thread context class loader is set to the initial, even in case of a failure`() {
+        assertThatThrownBy { executeWithThreadContextClassLoader(temporaryClassLoader) {
+            throw RuntimeException()
+        } }.isInstanceOf(RuntimeException::class.java)
+
+        assertThat(Thread.currentThread().contextClassLoader).isNotEqualTo(temporaryClassLoader)
     }
 
 }

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

@@ -40,6 +40,7 @@ import net.corda.node.internal.cordapp.*
 import net.corda.node.internal.rpc.proxies.AuthenticatedRpcOpsProxy
 import net.corda.node.internal.rpc.proxies.ExceptionMaskingRpcOpsProxy
 import net.corda.node.internal.rpc.proxies.ExceptionSerialisingRpcOpsProxy
+import net.corda.node.internal.rpc.proxies.ThreadContextAdjustingRpcOpsProxy
 import net.corda.node.services.ContractUpgradeHandler
 import net.corda.node.services.FinalityHandler
 import net.corda.node.services.NotaryChangeHandler
@@ -254,7 +255,7 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
     }
 
     /** The implementation of the [CordaRPCOps] interface used by this node. */
-    open fun makeRPCOps(): CordaRPCOps {
+    open fun makeRPCOps(cordappLoader: CordappLoader): CordaRPCOps {
         val ops: CordaRPCOps = CordaRPCOpsImpl(services, smm, flowStarter) { shutdownExecutor.submit { stop() } }.also { it.closeOnStop() }
         val proxies = mutableListOf<(CordaRPCOps) -> CordaRPCOps>()
         // Mind that order is relevant here.
@@ -263,6 +264,7 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
             proxies += { it -> ExceptionMaskingRpcOpsProxy(it, true) }
         }
         proxies += { it -> ExceptionSerialisingRpcOpsProxy(it, configuration.devMode) }
+        proxies += { it -> ThreadContextAdjustingRpcOpsProxy(it, cordappLoader.appClassLoader) }
         return proxies.fold(ops) { delegate, decorate -> decorate(delegate) }
     }
 
@@ -323,7 +325,7 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
         installCoreFlows()
         registerCordappFlows()
         services.rpcFlows += cordappLoader.cordapps.flatMap { it.rpcFlows }
-        val rpcOps = makeRPCOps()
+        val rpcOps = makeRPCOps(cordappLoader)
         startShell()
         networkMapClient?.start(trustRoot)
 

node/src/main/kotlin/net/corda/node/internal/rpc/proxies/ThreadContextAdjustingRpcOpsProxy.kt

@@ -0,0 +1,29 @@
+package net.corda.node.internal.rpc.proxies
+
+import net.corda.core.internal.executeWithThreadContextClassLoader
+import net.corda.core.messaging.CordaRPCOps
+import net.corda.node.internal.InvocationHandlerTemplate
+import java.lang.reflect.Method
+import java.lang.reflect.Proxy
+
+/**
+ * A [CordaRPCOps] proxy that adjusts the thread context's class loader temporarily on every invocation with the provided classloader.
+ * As an example, this can be used to work-around cases, where 3rd party libraries prioritise the thread context's class loader over the current one,
+ * without sensible fallbacks to the classloader of the current instance.
+ * If clients' CorDapps use one of these libraries, this temporary adjustment can ensure that any provided classes from these libraries will be available during RPC calls.
+ */
+internal class ThreadContextAdjustingRpcOpsProxy(private val delegate: CordaRPCOps, private val classLoader: ClassLoader): CordaRPCOps by proxy(delegate, classLoader) {
+    private companion object {
+        private fun proxy(delegate: CordaRPCOps, classLoader: ClassLoader): CordaRPCOps {
+            val handler = ThreadContextAdjustingRpcOpsProxy.ThreadContextAdjustingInvocationHandler(delegate, classLoader)
+            return Proxy.newProxyInstance(delegate::class.java.classLoader, arrayOf(CordaRPCOps::class.java), handler) as CordaRPCOps
+        }
+    }
+
+    private class ThreadContextAdjustingInvocationHandler(override val delegate: CordaRPCOps, private val classLoader: ClassLoader) : InvocationHandlerTemplate {
+        override fun invoke(proxy: Any, method: Method, arguments: Array<out Any?>?): Any? {
+            return executeWithThreadContextClassLoader(this.classLoader) { super.invoke(proxy, method, arguments) }
+        }
+    }
+
+}

node/src/test/kotlin/net/corda/node/internal/rpc/proxies/ThreadContextAdjustingRpcOpsProxyTest.kt

@@ -0,0 +1,32 @@
+package net.corda.node.internal.rpc.proxies
+
+import com.nhaarman.mockito_kotlin.any
+import com.nhaarman.mockito_kotlin.mock
+import net.corda.core.flows.StateMachineRunId
+import net.corda.core.messaging.CordaRPCOps
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.Test
+import org.mockito.Mockito.`when`
+
+class ThreadContextAdjustingRpcOpsProxyTest {
+
+    private val coreOps = mock<InstrumentedCordaRPCOps>()
+    private val mockClassloader = mock<ClassLoader>()
+    private val proxy = ThreadContextAdjustingRpcOpsProxy(coreOps, mockClassloader)
+
+
+    private interface InstrumentedCordaRPCOps: CordaRPCOps {
+        fun getThreadContextClassLoader(): ClassLoader = Thread.currentThread().contextClassLoader
+    }
+
+    @Test
+    fun verifyThreadContextIsAdjustedTemporarily() {
+        `when`(coreOps.killFlow(any())).thenAnswer {
+            assertThat(Thread.currentThread().contextClassLoader).isEqualTo(mockClassloader)
+            true
+        }
+
+        val result = proxy.killFlow(StateMachineRunId.createRandom())
+        assertThat(Thread.currentThread().contextClassLoader).isNotEqualTo(mockClassloader)
+    }
+}

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

@@ -7,24 +7,6 @@ import net.corda.core.serialization.SerializationCustomSerializer
 class CurrencySerializer : SerializationCustomSerializer<Currency, CurrencySerializer.Proxy> {
     data class Proxy(val currency: String)
 
-    override fun fromProxy(proxy: Proxy): Currency {
-        return withCurrentClassLoader { Currency.parse(proxy.currency) }
-    }
+    override fun fromProxy(proxy: Proxy): Currency = Currency.parse(proxy.currency)
     override fun toProxy(obj: Currency) = Proxy(obj.toString())
-
-    /**
-     * The initialization of [Currency] uses the classpath to identify needed resources.
-     * However, it gives priority to the classloader in the thread context over the one this class was loaded with.
-     * See: [com.opengamma.strata.collect.io.ResourceLocator.classLoader]
-     *
-     * This is the reason we temporarily override the class loader in the thread context here, with the classloader of this
-     * class, which is guaranteed to contain everything in the 3rd party library's classpath.
-     */
-    private fun withCurrentClassLoader(serializationFunction: () -> Currency): Currency {
-        val threadClassLoader = Thread.currentThread().contextClassLoader
-        Thread.currentThread().contextClassLoader = this.javaClass.classLoader
-        val result =serializationFunction()
-        Thread.currentThread().contextClassLoader = threadClassLoader
-        return result
-    }
 }