[CORDA-303]: Add some tests that stop and start a node and check state is persistent. (#1449)

* [CORDA-303]: Add some tests that stop and start a node and check state is persistent.

* [CORDA-303]: Updated change log and added Javadocs.

* [CORDA-303]: Cleaned up test.

* [CORDA-303]: Removed blank lines after class or function declarations.

* [CORDA-303]: Wrapped multiple invocations in `with` construct.

docs/source/changelog.rst

@@ -124,6 +124,8 @@ UNRELEASED
   directly build a ``FilteredTransaction`` using provided filtering functions, without first accessing the
   ``tx: WireTransaction``.
 
+* Test type ``NodeHandle`` now has method ``stop(): CordaFuture<Unit>`` that terminates the referenced node.
+
 Milestone 14
 ------------
 

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

@@ -0,0 +1,160 @@
+package net.corda.test.node
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.contracts.Command
+import net.corda.core.contracts.CommandData
+import net.corda.core.contracts.Contract
+import net.corda.core.contracts.LinearState
+import net.corda.core.contracts.UniqueIdentifier
+import net.corda.core.contracts.requireSingleCommand
+import net.corda.core.contracts.requireThat
+import net.corda.core.flows.FinalityFlow
+import net.corda.core.flows.FlowLogic
+import net.corda.core.flows.StartableByRPC
+import net.corda.core.identity.AbstractParty
+import net.corda.core.identity.Party
+import net.corda.core.messaging.startFlow
+import net.corda.core.node.services.ServiceInfo
+import net.corda.core.schemas.MappedSchema
+import net.corda.core.schemas.PersistentState
+import net.corda.core.schemas.QueryableState
+import net.corda.core.serialization.CordaSerializable
+import net.corda.core.transactions.LedgerTransaction
+import net.corda.core.transactions.SignedTransaction
+import net.corda.core.transactions.TransactionBuilder
+import net.corda.core.utilities.ProgressTracker
+import net.corda.core.utilities.getOrThrow
+import net.corda.node.services.FlowPermissions
+import net.corda.node.services.transactions.SimpleNotaryService
+import net.corda.nodeapi.User
+import net.corda.testing.DUMMY_NOTARY
+import net.corda.testing.driver.driver
+import org.junit.Test
+import java.lang.management.ManagementFactory
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.Table
+import kotlin.test.assertEquals
+
+class NodeStatePersistenceTests {
+
+    @Test
+    fun `persistent state survives node restart`() {
+        val user = User("mark", "dadada", setOf(FlowPermissions.startFlowPermission<SendMessageFlow>()))
+        val message = Message("Hello world!")
+        driver(isDebug = true, startNodesInProcess = isQuasarAgentSpecified()) {
+
+            startNode(providedName = DUMMY_NOTARY.name, advertisedServices = setOf(ServiceInfo(SimpleNotaryService.type))).getOrThrow()
+            var nodeHandle = startNode(rpcUsers = listOf(user)).getOrThrow()
+            val nodeName = nodeHandle.nodeInfo.legalIdentity.name
+            nodeHandle.rpcClientToNode().start(user.username, user.password).use {
+                it.proxy.startFlow(::SendMessageFlow, message).returnValue.getOrThrow()
+            }
+            nodeHandle.stop().getOrThrow()
+
+            nodeHandle = startNode(providedName = nodeName, rpcUsers = listOf(user)).getOrThrow()
+            nodeHandle.rpcClientToNode().start(user.username, user.password).use {
+                val page = it.proxy.vaultQuery(MessageState::class.java)
+                val retrievedMessage = page.states.singleOrNull()?.state?.data?.message
+                assertEquals(message, retrievedMessage)
+            }
+        }
+    }
+}
+
+fun isQuasarAgentSpecified(): Boolean {
+    val jvmArgs = ManagementFactory.getRuntimeMXBean().inputArguments
+    return jvmArgs.any { it.startsWith("-javaagent:") && it.endsWith("quasar.jar") }
+}
+
+@CordaSerializable
+data class Message(val value: String)
+
+data class MessageState(val message: Message, val by: Party, override val linearId: UniqueIdentifier = UniqueIdentifier()) : LinearState, QueryableState {
+    override val contract = MessageContract()
+    override val participants: List<AbstractParty> = listOf(by)
+
+    override fun generateMappedObject(schema: MappedSchema): PersistentState {
+        return when (schema) {
+            is MessageSchemaV1 -> MessageSchemaV1.PersistentMessage(
+                    by = by.name.toString(),
+                    value = message.value
+            )
+            else -> throw IllegalArgumentException("Unrecognised schema $schema")
+        }
+    }
+
+    override fun supportedSchemas(): Iterable<MappedSchema> = listOf(MessageSchemaV1)
+}
+
+object MessageSchema
+object MessageSchemaV1 : MappedSchema(
+        schemaFamily = MessageSchema.javaClass,
+        version = 1,
+        mappedTypes = listOf(PersistentMessage::class.java)) {
+
+    @Entity
+    @Table(name = "messages")
+    class PersistentMessage(
+            @Column(name = "by")
+            var by: String,
+
+            @Column(name = "value")
+            var value: String
+    ) : PersistentState()
+}
+
+open class MessageContract : Contract {
+    override fun verify(tx: LedgerTransaction) {
+        val command = tx.commands.requireSingleCommand<Commands.Send>()
+        requireThat {
+            // Generic constraints around the IOU transaction.
+            "No inputs should be consumed when sending a message." using (tx.inputs.isEmpty())
+            "Only one output state should be created." using (tx.outputs.size == 1)
+            val out = tx.outputsOfType<MessageState>().single()
+            "Message sender must sign." using (command.signers.containsAll(out.participants.map { it.owningKey }))
+
+            "Message value must not be empty." using (out.message.value.isNotBlank())
+        }
+    }
+
+    interface Commands : CommandData {
+        class Send : Commands
+    }
+}
+
+@StartableByRPC
+class SendMessageFlow(private val message: Message) : FlowLogic<SignedTransaction>() {
+    companion object {
+        object GENERATING_TRANSACTION : ProgressTracker.Step("Generating transaction based on the message.")
+        object VERIFYING_TRANSACTION : ProgressTracker.Step("Verifying contract constraints.")
+        object SIGNING_TRANSACTION : ProgressTracker.Step("Signing transaction with our private key.")
+        object FINALISING_TRANSACTION : ProgressTracker.Step("Obtaining notary signature and recording transaction.") {
+            override fun childProgressTracker() = FinalityFlow.tracker()
+        }
+
+        fun tracker() = ProgressTracker(GENERATING_TRANSACTION, VERIFYING_TRANSACTION, SIGNING_TRANSACTION, FINALISING_TRANSACTION)
+    }
+
+    override val progressTracker = tracker()
+
+    @Suspendable
+    override fun call(): SignedTransaction {
+        val notary = serviceHub.networkMapCache.getAnyNotary()
+
+        progressTracker.currentStep = GENERATING_TRANSACTION
+
+        val messageState = MessageState(message = message, by = serviceHub.myInfo.legalIdentity)
+        val txCommand = Command(MessageContract.Commands.Send(), messageState.participants.map { it.owningKey })
+        val txBuilder = TransactionBuilder(notary).withItems(messageState, txCommand)
+
+        progressTracker.currentStep = VERIFYING_TRANSACTION
+        txBuilder.toWireTransaction().toLedgerTransaction(serviceHub).verify()
+
+        progressTracker.currentStep = SIGNING_TRANSACTION
+        val signedTx = serviceHub.signInitialTransaction(txBuilder)
+
+        progressTracker.currentStep = FINALISING_TRANSACTION
+        return subFlow(FinalityFlow(signedTx, FINALISING_TRANSACTION.childProgressTracker())).single()
+    }
+}

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

@@ -188,7 +188,15 @@ sealed class NodeHandle {
             override val webAddress: NetworkHostAndPort,
             val debugPort: Int?,
             val process: Process
-    ) : NodeHandle()
+    ) : NodeHandle() {
+        override fun stop(): CordaFuture<Unit> {
+            with(process) {
+                destroy()
+                waitFor()
+            }
+            return doneFuture(Unit)
+        }
+    }
 
     data class InProcess(
             override val nodeInfo: NodeInfo,
@@ -197,9 +205,23 @@ sealed class NodeHandle {
             override val webAddress: NetworkHostAndPort,
             val node: Node,
             val nodeThread: Thread
-    ) : NodeHandle()
+    ) : NodeHandle() {
+        override fun stop(): CordaFuture<Unit> {
+            node.stop()
+            with(nodeThread) {
+                interrupt()
+                join()
+            }
+            return doneFuture(Unit)
+        }
+    }
 
     fun rpcClientToNode(): CordaRPCClient = CordaRPCClient(configuration.rpcAddress!!, initialiseSerialization = false)
+
+    /**
+     * Stops the referenced node.
+     */
+    abstract fun stop(): CordaFuture<Unit>
 }
 
 data class WebserverHandle(