CORDA-299: Remove progress Observable from FlowHandle, unless explicitly requested. (#513)

* Remove progress Observable from FlowHandle, unless explicitly requested.
* Refactor FlowHandle creation into FlowStateMachine.
* Prevent server-side queue subscription for dummy Observable.
* Refactor so that RPC client does not receive any unused progress Observables. This is the simplest way of ensuring we have no dangling "hot" Observables when the RPC client closes.
* Test flow has correct handle.
* Resolve some compiler warnings.
* Document how starting a flow does not involve progress tracking by default.
* Update changelog and release notes for RPC API.
* Rename new RPC API to startTrackedFlow().
* Remove optimisation because of its affect on the client-side.
* Update documentation.

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

@@ -4,14 +4,10 @@ import net.corda.core.contracts.Amount
 import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.UpgradedContract
-import net.corda.core.crypto.Party
 import net.corda.core.crypto.SecureHash
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.StateMachineRunId
-import net.corda.core.messaging.CordaRPCOps
-import net.corda.core.messaging.FlowHandle
-import net.corda.core.messaging.StateMachineInfo
-import net.corda.core.messaging.StateMachineUpdate
+import net.corda.core.messaging.*
 import net.corda.core.node.NodeInfo
 import net.corda.core.node.services.NetworkMapCache
 import net.corda.core.node.services.StateMachineTransactionMapping
@@ -20,7 +16,6 @@ import net.corda.core.transactions.SignedTransaction
 import net.corda.node.services.api.ServiceHubInternal
 import net.corda.node.services.messaging.requirePermission
 import net.corda.node.services.startFlowPermission
-import net.corda.node.services.statemachine.FlowStateMachineImpl
 import net.corda.node.services.statemachine.StateMachineManager
 import net.corda.node.utilities.AddOrRemove
 import net.corda.node.utilities.transaction
@@ -41,7 +36,7 @@ class CordaRPCOpsImpl(
         private val smm: StateMachineManager,
         private val database: Database
 ) : CordaRPCOps {
-    override val protocolVersion: Int get() = 0
+    override val protocolVersion: Int = 0
 
     override fun networkMapUpdates(): Pair<List<NodeInfo>, Observable<NetworkMapCache.MapChange>> {
         return database.transaction {
@@ -100,15 +95,16 @@ class CordaRPCOpsImpl(
         }
     }
 
+    // TODO: Check that this flow is annotated as being intended for RPC invocation
+    override fun <T : Any> startTrackedFlowDynamic(logicType: Class<out FlowLogic<T>>, vararg args: Any?): FlowProgressHandle<T> {
+        requirePermission(startFlowPermission(logicType))
+        return services.invokeFlowAsync(logicType, *args).createHandle(hasProgress = true) as FlowProgressHandle<T>
+    }
+
     // TODO: Check that this flow is annotated as being intended for RPC invocation
     override fun <T : Any> startFlowDynamic(logicType: Class<out FlowLogic<T>>, vararg args: Any?): FlowHandle<T> {
         requirePermission(startFlowPermission(logicType))
-        val stateMachine = services.invokeFlowAsync(logicType, *args) as FlowStateMachineImpl<T>
-        return FlowHandle(
-                id = stateMachine.id,
-                progress = stateMachine.logic.track()?.second ?: Observable.empty(),
-                returnValue = stateMachine.resultFuture
-        )
+        return services.invokeFlowAsync(logicType, *args).createHandle(hasProgress = false)
     }
 
     override fun attachmentExists(id: SecureHash): Boolean {

node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt

@@ -6,6 +6,7 @@ import co.paralleluniverse.fibers.Suspendable
 import co.paralleluniverse.strands.Strand
 import com.google.common.util.concurrent.ListenableFuture
 import com.google.common.util.concurrent.SettableFuture
+import net.corda.client.rpc.notUsed
 import net.corda.core.abbreviate
 import net.corda.core.crypto.Party
 import net.corda.core.crypto.SecureHash
@@ -13,7 +14,10 @@ import net.corda.core.flows.FlowException
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.FlowStateMachine
 import net.corda.core.flows.StateMachineRunId
+import net.corda.core.messaging.FlowHandle
+import net.corda.core.messaging.FlowProgressHandle
 import net.corda.core.random63BitValue
+import net.corda.core.serialization.CordaSerializable
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.UntrustworthyData
@@ -27,6 +31,7 @@ import org.jetbrains.exposed.sql.Transaction
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
+import rx.Observable
 import java.sql.Connection
 import java.sql.SQLException
 import java.util.*
@@ -98,6 +103,15 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
         logic.stateMachine = this
     }
 
+    override fun createHandle(hasProgress: Boolean): FlowHandle<R> = if (hasProgress)
+        FlowProgressHandleImpl(
+            id = id,
+            returnValue = resultFuture,
+            progress = logic.track()?.second ?: Observable.empty()
+        )
+    else
+        FlowHandleImpl(id = id, returnValue = resultFuture)
+
     @Suspendable
     override fun run() {
         createTransaction()
@@ -410,3 +424,35 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
         timer.update(duration, TimeUnit.NANOSECONDS)
     }
 }
+
+// I would prefer for [FlowProgressHandleImpl] to extend [FlowHandleImpl],
+// but Kotlin doesn't allow this for data classes, not even to create
+// another data class!
+@CordaSerializable
+private data class FlowHandleImpl<A>(
+        override val id: StateMachineRunId,
+        override val returnValue: ListenableFuture<A>) : FlowHandle<A> {
+
+    /**
+     * Use this function for flows whose returnValue is not going to be used, so as to free up server resources.
+     */
+    override fun close() {
+        returnValue.cancel(false)
+    }
+}
+
+@CordaSerializable
+private data class FlowProgressHandleImpl<A> (
+        override val id: StateMachineRunId,
+        override val returnValue: ListenableFuture<A>,
+        override val progress: Observable<String>) : FlowProgressHandle<A> {
+
+    /**
+     * Use this function for flows whose returnValue and progress are not going to be used or tracked, so as to free up server resources.
+     * Note that it won't really close if one subscribes on progress [Observable], but then forgets to unsubscribe.
+     */
+    override fun close() {
+        progress.notUsed()
+        returnValue.cancel(false)
+    }
+}

node/src/test/kotlin/net/corda/node/CordaRPCOpsImplTest.kt

@@ -29,13 +29,12 @@ import net.corda.testing.node.MockNetwork.MockNode
 import net.corda.testing.sequence
 import org.apache.commons.io.IOUtils
 import org.assertj.core.api.Assertions.assertThatExceptionOfType
+import org.junit.Assert.assertArrayEquals
 import org.junit.Before
 import org.junit.Test
 import rx.Observable
 import java.io.ByteArrayOutputStream
-import java.util.*
-import kotlin.test.assertEquals
-import kotlin.test.assertFalse
+import kotlin.test.*
 
 class CordaRPCOpsImplTest {
 
@@ -210,7 +209,7 @@ class CordaRPCOpsImplTest {
     fun `can upload an attachment`() {
         val inputJar = Thread.currentThread().contextClassLoader.getResourceAsStream(testJar)
         val secureHash = rpc.uploadAttachment(inputJar)
-        assert(rpc.attachmentExists(secureHash))
+        assertTrue(rpc.attachmentExists(secureHash))
     }
 
     @Test
@@ -223,6 +222,6 @@ class CordaRPCOpsImplTest {
         IOUtils.copy(Thread.currentThread().contextClassLoader.getResourceAsStream(testJar), bufferFile)
         IOUtils.copy(rpc.openAttachment(secureHash), bufferRpc)
 
-        assert(Arrays.equals(bufferFile.toByteArray(), bufferRpc.toByteArray()))
+        assertArrayEquals(bufferFile.toByteArray(), bufferRpc.toByteArray())
     }
 }

node/src/test/kotlin/net/corda/node/InteractiveShellTest.kt

@@ -8,6 +8,7 @@ import net.corda.core.crypto.SecureHash
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.FlowStateMachine
 import net.corda.core.flows.StateMachineRunId
+import net.corda.core.messaging.FlowProgressHandle
 import net.corda.core.node.ServiceHub
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.DUMMY_PUBKEY_1
@@ -84,6 +85,8 @@ class InteractiveShellTest {
             throw UnsupportedOperationException("not implemented")
         }
 
+        override fun createHandle(hasProgress: Boolean): FlowProgressHandle<Any?> = throw UnsupportedOperationException("not implemented")
+
         override val serviceHub: ServiceHub
             get() = throw UnsupportedOperationException()
         override val logger: Logger