CORDA-3677 FlowExternalOperation serialising reference to FlowLogic (#6094)

* Stop capturing 'FlowLogic' references in flowAsyncOperation;

Creating concrete classes removes the implicit reference to FlowLogic (as this) being included in the anonymous object

* Modify test code so that lambdas no longer get implicit references to their enclosing 'FlowLogic'

core-tests/src/test/kotlin/net/corda/coretests/flows/FlowExternalAsyncOperationTest.kt

@@ -177,12 +177,14 @@ class FlowExternalAsyncOperationTest : AbstractFlowExternalOperationTest() {
         FlowWithExternalProcess(party) {
 
         @Suspendable
-        override fun testCode(): Any =
+        override fun testCode(): Any {
-            await(ExternalAsyncOperation(serviceHub) { _, _ ->
+            val e = createException()
+            return await(ExternalAsyncOperation(serviceHub) { _, _ ->
                 CompletableFuture<Any>().apply {
-                    completeExceptionally(createException())
+                    completeExceptionally(e)
                 }
             })
+        }
 
         private fun createException() = when (exceptionType) {
             HospitalizeFlowException::class.java -> HospitalizeFlowException("keep it around")

core-tests/src/test/kotlin/net/corda/coretests/flows/FlowExternalOperationTest.kt

@@ -235,7 +235,10 @@ class FlowExternalOperationTest : AbstractFlowExternalOperationTest() {
         FlowWithExternalProcess(party) {
 
         @Suspendable
-        override fun testCode(): Any = await(ExternalOperation(serviceHub) { _, _ -> throw createException() })
+        override fun testCode() {
+            val e = createException()
+            await(ExternalOperation(serviceHub) { _, _ -> throw e })
+        }
 
         private fun createException() = when (exceptionType) {
             HospitalizeFlowException::class.java -> HospitalizeFlowException("keep it around")

core/src/main/kotlin/net/corda/core/flows/FlowLogic.kt

@@ -565,12 +565,7 @@ abstract class FlowLogic<out T> {
     @Suspendable
     fun <R : Any> await(operation: FlowExternalAsyncOperation<R>): R {
         // Wraps the passed in [FlowExternalAsyncOperation] so its [CompletableFuture] can be converted into a [CordaFuture]
-        val flowAsyncOperation = object : FlowAsyncOperation<R>, WrappedFlowExternalAsyncOperation<R> {
+        val flowAsyncOperation = WrappedFlowExternalAsyncOperation(operation)
-            override val operation = operation
-            override fun execute(deduplicationId: String): CordaFuture<R> {
-                return this.operation.execute(deduplicationId).asCordaFuture()
-            }
-        }
         val request = FlowIORequest.ExecuteAsyncOperation(flowAsyncOperation)
         return stateMachine.suspend(request, false)
     }
@@ -584,18 +579,7 @@ abstract class FlowLogic<out T> {
      */
     @Suspendable
     fun <R : Any> await(operation: FlowExternalOperation<R>): R {
-        val flowAsyncOperation = object : FlowAsyncOperation<R>, WrappedFlowExternalOperation<R> {
+        val flowAsyncOperation = WrappedFlowExternalOperation(serviceHub as ServiceHubCoreInternal, operation)
-            override val serviceHub = this@FlowLogic.serviceHub as ServiceHubCoreInternal
-            override val operation = operation
-            override fun execute(deduplicationId: String): CordaFuture<R> {
-                // Using a [CompletableFuture] allows unhandled exceptions to be thrown inside the background operation
-                // the exceptions will be set on the future by [CompletableFuture.AsyncSupply.run]
-                return CompletableFuture.supplyAsync(
-                    Supplier { this.operation.execute(deduplicationId) },
-                    serviceHub.externalOperationExecutor
-                ).asCordaFuture()
-            }
-        }
         val request = FlowIORequest.ExecuteAsyncOperation(flowAsyncOperation)
         return stateMachine.suspend(request, false)
     }
@@ -605,21 +589,32 @@ abstract class FlowLogic<out T> {
  * [WrappedFlowExternalAsyncOperation] is added to allow jackson to properly reference the data stored within the wrapped
  * [FlowExternalAsyncOperation].
  */
-private interface WrappedFlowExternalAsyncOperation<R : Any> {
+private class WrappedFlowExternalAsyncOperation<R : Any>(val operation: FlowExternalAsyncOperation<R>) : FlowAsyncOperation<R> {
-    val operation: FlowExternalAsyncOperation<R>
+    override fun execute(deduplicationId: String): CordaFuture<R> {
+        return operation.execute(deduplicationId).asCordaFuture()
+    }
 }
 
 /**
  * [WrappedFlowExternalOperation] is added to allow jackson to properly reference the data stored within the wrapped
  * [FlowExternalOperation].
  *
- * The reference to [ServiceHub] is is also needed by Kryo to properly keep a reference to [ServiceHub] so that
+ * The reference to [ServiceHub] is also needed by Kryo to properly keep a reference to [ServiceHub] so that
  * [FlowExternalOperation] can be run from the [ServiceHubCoreInternal.externalOperationExecutor] without causing errors when retrying a
  * flow. A [NullPointerException] is thrown if [FlowLogic.serviceHub] is accessed from [FlowLogic.await] when retrying a flow.
  */
-private interface WrappedFlowExternalOperation<R : Any> {
+private class WrappedFlowExternalOperation<R : Any>(
-    val serviceHub: ServiceHub
+    val serviceHub: ServiceHubCoreInternal,
     val operation: FlowExternalOperation<R>
+) : FlowAsyncOperation<R> {
+    override fun execute(deduplicationId: String): CordaFuture<R> {
+        // Using a [CompletableFuture] allows unhandled exceptions to be thrown inside the background operation
+        // the exceptions will be set on the future by [CompletableFuture.AsyncSupply.run]
+        return CompletableFuture.supplyAsync(
+            Supplier { this.operation.execute(deduplicationId) },
+            serviceHub.externalOperationExecutor
+        ).asCordaFuture()
+    }
 }
 
 /**