Various structural cleanups of node-driver:

* Extracted out ShutdownManager into its own file
* Moved RPCDriver and ProcessUtilities into internal package
* Made n.c.testing.performance package internal
This commit is contained in:
Shams Asari 2017-11-06 20:41:06 +00:00
parent e6feca2f03
commit e26e41a384
21 changed files with 152 additions and 131 deletions

View File

@ -11,8 +11,8 @@ import net.corda.core.serialization.serialize
import net.corda.core.utilities.* import net.corda.core.utilities.*
import net.corda.node.services.messaging.RPCServerConfiguration import net.corda.node.services.messaging.RPCServerConfiguration
import net.corda.nodeapi.RPCApi import net.corda.nodeapi.RPCApi
import net.corda.testing.*
import net.corda.testing.driver.poll import net.corda.testing.driver.poll
import net.corda.testing.internal.*
import org.apache.activemq.artemis.api.core.SimpleString import org.apache.activemq.artemis.api.core.SimpleString
import org.junit.Assert.assertEquals import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue import org.junit.Assert.assertTrue

View File

@ -6,10 +6,10 @@ import net.corda.core.internal.concurrent.map
import net.corda.core.messaging.RPCOps import net.corda.core.messaging.RPCOps
import net.corda.node.services.messaging.RPCServerConfiguration import net.corda.node.services.messaging.RPCServerConfiguration
import net.corda.nodeapi.User import net.corda.nodeapi.User
import net.corda.testing.RPCDriverExposedDSLInterface import net.corda.testing.internal.RPCDriverExposedDSLInterface
import net.corda.testing.rpcTestUser import net.corda.testing.internal.rpcTestUser
import net.corda.testing.startInVmRpcClient import net.corda.testing.internal.startInVmRpcClient
import net.corda.testing.startRpcClient import net.corda.testing.internal.startRpcClient
import org.apache.activemq.artemis.api.core.client.ClientSession import org.apache.activemq.artemis.api.core.client.ClientSession
import org.junit.runners.Parameterized import org.junit.runners.Parameterized

View File

@ -7,9 +7,9 @@ import net.corda.core.internal.concurrent.thenMatch
import net.corda.core.messaging.RPCOps import net.corda.core.messaging.RPCOps
import net.corda.core.utilities.getOrThrow import net.corda.core.utilities.getOrThrow
import net.corda.node.services.messaging.rpcContext import net.corda.node.services.messaging.rpcContext
import net.corda.testing.RPCDriverExposedDSLInterface import net.corda.testing.internal.RPCDriverExposedDSLInterface
import net.corda.testing.rpcDriver import net.corda.testing.internal.rpcDriver
import net.corda.testing.rpcTestUser import net.corda.testing.internal.rpcTestUser
import org.assertj.core.api.Assertions.assertThat import org.assertj.core.api.Assertions.assertThat
import org.junit.Test import org.junit.Test
import org.junit.runner.RunWith import org.junit.runner.RunWith

View File

@ -7,8 +7,8 @@ import net.corda.core.crypto.random63BitValue
import net.corda.core.internal.concurrent.fork import net.corda.core.internal.concurrent.fork
import net.corda.core.serialization.CordaSerializable import net.corda.core.serialization.CordaSerializable
import net.corda.node.services.messaging.RPCServerConfiguration import net.corda.node.services.messaging.RPCServerConfiguration
import net.corda.testing.RPCDriverExposedDSLInterface import net.corda.testing.internal.RPCDriverExposedDSLInterface
import net.corda.testing.rpcDriver import net.corda.testing.internal.rpcDriver
import org.junit.Test import org.junit.Test
import org.junit.runner.RunWith import org.junit.runner.RunWith
import org.junit.runners.Parameterized import org.junit.runners.Parameterized

View File

@ -5,8 +5,8 @@ import net.corda.core.concurrent.CordaFuture
import net.corda.core.internal.concurrent.openFuture import net.corda.core.internal.concurrent.openFuture
import net.corda.core.messaging.* import net.corda.core.messaging.*
import net.corda.core.utilities.getOrThrow import net.corda.core.utilities.getOrThrow
import net.corda.testing.rpcDriver import net.corda.testing.internal.rpcDriver
import net.corda.testing.startRpcClient import net.corda.testing.internal.startRpcClient
import org.assertj.core.api.Assertions.assertThatThrownBy import org.assertj.core.api.Assertions.assertThatThrownBy
import org.junit.Test import org.junit.Test

View File

@ -5,14 +5,14 @@ import net.corda.client.rpc.internal.RPCClientConfiguration
import net.corda.core.messaging.RPCOps import net.corda.core.messaging.RPCOps
import net.corda.core.utilities.minutes import net.corda.core.utilities.minutes
import net.corda.core.utilities.seconds import net.corda.core.utilities.seconds
import net.corda.testing.performance.div import net.corda.testing.internal.performance.div
import net.corda.node.services.messaging.RPCServerConfiguration import net.corda.node.services.messaging.RPCServerConfiguration
import net.corda.testing.RPCDriverExposedDSLInterface import net.corda.testing.internal.RPCDriverExposedDSLInterface
import net.corda.testing.measure import net.corda.testing.measure
import net.corda.testing.performance.startPublishingFixedRateInjector import net.corda.testing.internal.performance.startPublishingFixedRateInjector
import net.corda.testing.performance.startReporter import net.corda.testing.internal.performance.startReporter
import net.corda.testing.performance.startTightLoopInjector import net.corda.testing.internal.performance.startTightLoopInjector
import net.corda.testing.rpcDriver import net.corda.testing.internal.rpcDriver
import org.junit.Ignore import org.junit.Ignore
import org.junit.Test import org.junit.Test
import org.junit.runner.RunWith import org.junit.runner.RunWith

View File

@ -6,8 +6,8 @@ import net.corda.node.services.Permissions.Companion.invokeRpc
import net.corda.node.services.messaging.rpcContext import net.corda.node.services.messaging.rpcContext
import net.corda.node.services.messaging.requirePermission import net.corda.node.services.messaging.requirePermission
import net.corda.nodeapi.User import net.corda.nodeapi.User
import net.corda.testing.RPCDriverExposedDSLInterface import net.corda.testing.internal.RPCDriverExposedDSLInterface
import net.corda.testing.rpcDriver import net.corda.testing.internal.rpcDriver
import org.junit.Test import org.junit.Test
import org.junit.runner.RunWith import org.junit.runner.RunWith
import org.junit.runners.Parameterized import org.junit.runners.Parameterized

View File

@ -23,6 +23,10 @@ import net.corda.nodeapi.User
import net.corda.testing.* import net.corda.testing.*
import net.corda.testing.contracts.DummyContract import net.corda.testing.contracts.DummyContract
import net.corda.testing.contracts.DummyContractV2 import net.corda.testing.contracts.DummyContractV2
import net.corda.testing.internal.RPCDriverExposedDSLInterface
import net.corda.testing.internal.rpcDriver
import net.corda.testing.internal.rpcTestUser
import net.corda.testing.internal.startRpcClient
import net.corda.testing.node.MockNetwork import net.corda.testing.node.MockNetwork
import org.junit.After import org.junit.After
import org.junit.Before import org.junit.Before

View File

@ -569,7 +569,7 @@ class ObligationTests {
// Try defaulting an obligation due in the future // Try defaulting an obligation due in the future
val pastTestTime = TEST_TX_TIME - 7.days val pastTestTime = TEST_TX_TIME - 7.days
val futureTestTime = TEST_TX_TIME + 7.days val futureTestTime = TEST_TX_TIME + 7.days
transaction("Settlement") { transaction {
attachments(Obligation.PROGRAM_ID) attachments(Obligation.PROGRAM_ID)
input(Obligation.PROGRAM_ID, oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` futureTestTime) input(Obligation.PROGRAM_ID, oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` futureTestTime)
output(Obligation.PROGRAM_ID, "Alice's defaulted $1,000,000 obligation to Bob") { (oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` futureTestTime).copy(lifecycle = Lifecycle.DEFAULTED) } output(Obligation.PROGRAM_ID, "Alice's defaulted $1,000,000 obligation to Bob") { (oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` futureTestTime).copy(lifecycle = Lifecycle.DEFAULTED) }
@ -580,7 +580,7 @@ class ObligationTests {
// Try defaulting an obligation that is now in the past // Try defaulting an obligation that is now in the past
ledger { ledger {
transaction("Settlement") { transaction {
attachments(Obligation.PROGRAM_ID) attachments(Obligation.PROGRAM_ID)
input(Obligation.PROGRAM_ID, oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` pastTestTime) input(Obligation.PROGRAM_ID, oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` pastTestTime)
output(Obligation.PROGRAM_ID, "Alice's defaulted $1,000,000 obligation to Bob") { (oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` pastTestTime).copy(lifecycle = Lifecycle.DEFAULTED) } output(Obligation.PROGRAM_ID, "Alice's defaulted $1,000,000 obligation to Bob") { (oneMillionDollars.OBLIGATION between Pair(ALICE, BOB) `at` pastTestTime).copy(lifecycle = Lifecycle.DEFAULTED) }

View File

@ -18,10 +18,10 @@ import net.corda.testing.DUMMY_NOTARY
import net.corda.testing.driver.NodeHandle import net.corda.testing.driver.NodeHandle
import net.corda.testing.driver.driver import net.corda.testing.driver.driver
import net.corda.testing.node.NotarySpec import net.corda.testing.node.NotarySpec
import net.corda.testing.performance.div import net.corda.testing.internal.performance.div
import net.corda.testing.performance.startPublishingFixedRateInjector import net.corda.testing.internal.performance.startPublishingFixedRateInjector
import net.corda.testing.performance.startReporter import net.corda.testing.internal.performance.startReporter
import net.corda.testing.performance.startTightLoopInjector import net.corda.testing.internal.performance.startTightLoopInjector
import org.junit.Before import org.junit.Before
import org.junit.Ignore import org.junit.Ignore
import org.junit.Test import org.junit.Test

View File

@ -6,6 +6,7 @@ import net.corda.core.internal.concurrent.fork
import net.corda.core.internal.concurrent.map import net.corda.core.internal.concurrent.map
import net.corda.core.utilities.loggerFor import net.corda.core.utilities.loggerFor
import net.corda.testing.driver.* import net.corda.testing.driver.*
import net.corda.testing.internal.ProcessUtilities
import net.corda.testing.node.NotarySpec import net.corda.testing.node.NotarySpec
import okhttp3.OkHttpClient import okhttp3.OkHttpClient
import okhttp3.Request import okhttp3.Request

View File

@ -1,4 +1,4 @@
package net.corda.testing.driver package net.corda.testing.internal
import org.apache.commons.io.FileUtils import org.apache.commons.io.FileUtils
import org.junit.Rule import org.junit.Rule

View File

@ -45,7 +45,6 @@ fun ledger(
*/ */
@JvmOverloads @JvmOverloads
fun transaction( fun transaction(
transactionLabel: String? = null,
transactionBuilder: TransactionBuilder = TransactionBuilder(notary = DUMMY_NOTARY), transactionBuilder: TransactionBuilder = TransactionBuilder(notary = DUMMY_NOTARY),
initialiseSerialization: Boolean = true, initialiseSerialization: Boolean = true,
cordappPackages: List<String> = emptyList(), cordappPackages: List<String> = emptyList(),

View File

@ -37,6 +37,7 @@ import net.corda.nodeapi.internal.addShutdownHook
import net.corda.testing.* import net.corda.testing.*
import net.corda.testing.common.internal.NetworkParametersCopier import net.corda.testing.common.internal.NetworkParametersCopier
import net.corda.testing.common.internal.testNetworkParameters import net.corda.testing.common.internal.testNetworkParameters
import net.corda.testing.internal.ProcessUtilities
import net.corda.testing.node.ClusterSpec import net.corda.testing.node.ClusterSpec
import net.corda.testing.node.MockServices.Companion.MOCK_VERSION_INFO import net.corda.testing.node.MockServices.Companion.MOCK_VERSION_INFO
import net.corda.testing.node.NotarySpec import net.corda.testing.node.NotarySpec
@ -54,12 +55,10 @@ import java.time.Instant
import java.time.ZoneOffset.UTC import java.time.ZoneOffset.UTC
import java.time.format.DateTimeFormatter import java.time.format.DateTimeFormatter
import java.util.* import java.util.*
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit.MILLISECONDS import java.util.concurrent.TimeUnit.MILLISECONDS
import java.util.concurrent.TimeUnit.SECONDS import java.util.concurrent.TimeUnit.SECONDS
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicInteger
import kotlin.collections.ArrayList import kotlin.collections.ArrayList
import kotlin.concurrent.thread import kotlin.concurrent.thread
@ -559,102 +558,6 @@ fun <A> poll(
return resultFuture return resultFuture
} }
class ShutdownManager(private val executorService: ExecutorService) {
private class State {
val registeredShutdowns = ArrayList<CordaFuture<() -> Unit>>()
var isShutdown = false
}
private val state = ThreadBox(State())
companion object {
inline fun <A> run(providedExecutorService: ExecutorService? = null, block: ShutdownManager.() -> A): A {
val executorService = providedExecutorService ?: Executors.newScheduledThreadPool(1)
val shutdownManager = ShutdownManager(executorService)
try {
return block(shutdownManager)
} finally {
shutdownManager.shutdown()
providedExecutorService ?: executorService.shutdown()
}
}
}
fun shutdown() {
val shutdownActionFutures = state.locked {
if (isShutdown) {
emptyList<CordaFuture<() -> Unit>>()
} else {
isShutdown = true
registeredShutdowns
}
}
val shutdowns = shutdownActionFutures.map { Try.on { it.getOrThrow(1.seconds) } }
shutdowns.reversed().forEach {
when (it) {
is Try.Success ->
try {
it.value()
} catch (t: Throwable) {
log.warn("Exception while shutting down", t)
}
is Try.Failure -> log.warn("Exception while getting shutdown method, disregarding", it.exception)
}
}
}
fun registerShutdown(shutdown: CordaFuture<() -> Unit>) {
state.locked {
require(!isShutdown)
registeredShutdowns.add(shutdown)
}
}
fun registerShutdown(shutdown: () -> Unit) = registerShutdown(doneFuture(shutdown))
fun registerProcessShutdown(processFuture: CordaFuture<Process>) {
val processShutdown = processFuture.map { process ->
{
process.destroy()
/** Wait 5 seconds, then [Process.destroyForcibly] */
val finishedFuture = executorService.submit {
process.waitFor()
}
try {
finishedFuture.get(5, SECONDS)
} catch (exception: TimeoutException) {
finishedFuture.cancel(true)
process.destroyForcibly()
}
Unit
}
}
registerShutdown(processShutdown)
}
interface Follower {
fun unfollow()
fun shutdown()
}
fun follower() = object : Follower {
private val start = state.locked { registeredShutdowns.size }
private val end = AtomicInteger(start - 1)
override fun unfollow() = end.set(state.locked { registeredShutdowns.size })
override fun shutdown() = end.get().let { end ->
start > end && throw IllegalStateException("You haven't called unfollow.")
state.locked {
registeredShutdowns.subList(start, end).listIterator(end - start).run {
while (hasPrevious()) {
previous().getOrThrow().invoke()
set(doneFuture {}) // Don't break other followers by doing a remove.
}
}
}
}
}
}
class DriverDSL( class DriverDSL(
val portAllocation: PortAllocation, val portAllocation: PortAllocation,
val debugPortAllocation: PortAllocation, val debugPortAllocation: PortAllocation,

View File

@ -0,0 +1,113 @@
package net.corda.testing.driver
import net.corda.core.concurrent.CordaFuture
import net.corda.core.internal.ThreadBox
import net.corda.core.internal.concurrent.doneFuture
import net.corda.core.internal.concurrent.map
import net.corda.core.utilities.Try
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.loggerFor
import net.corda.core.utilities.seconds
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicInteger
class ShutdownManager(private val executorService: ExecutorService) {
private class State {
val registeredShutdowns = ArrayList<CordaFuture<() -> Unit>>()
var isShutdown = false
}
private val state = ThreadBox(State())
companion object {
private val log = loggerFor<ShutdownManager>()
inline fun <A> run(providedExecutorService: ExecutorService? = null, block: ShutdownManager.() -> A): A {
val executorService = providedExecutorService ?: Executors.newScheduledThreadPool(1)
val shutdownManager = ShutdownManager(executorService)
try {
return block(shutdownManager)
} finally {
shutdownManager.shutdown()
providedExecutorService ?: executorService.shutdown()
}
}
}
fun shutdown() {
val shutdownActionFutures = state.locked {
if (isShutdown) {
emptyList<CordaFuture<() -> Unit>>()
} else {
isShutdown = true
registeredShutdowns
}
}
val shutdowns = shutdownActionFutures.map { Try.on { it.getOrThrow(1.seconds) } }
shutdowns.reversed().forEach {
when (it) {
is Try.Success ->
try {
it.value()
} catch (t: Throwable) {
log.warn("Exception while shutting down", t)
}
is Try.Failure -> log.warn("Exception while getting shutdown method, disregarding", it.exception)
}
}
}
fun registerShutdown(shutdown: CordaFuture<() -> Unit>) {
state.locked {
require(!isShutdown)
registeredShutdowns.add(shutdown)
}
}
fun registerShutdown(shutdown: () -> Unit) = registerShutdown(doneFuture(shutdown))
fun registerProcessShutdown(processFuture: CordaFuture<Process>) {
val processShutdown = processFuture.map { process ->
{
process.destroy()
/** Wait 5 seconds, then [Process.destroyForcibly] */
val finishedFuture = executorService.submit {
process.waitFor()
}
try {
finishedFuture.get(5, TimeUnit.SECONDS)
} catch (exception: TimeoutException) {
finishedFuture.cancel(true)
process.destroyForcibly()
}
Unit
}
}
registerShutdown(processShutdown)
}
interface Follower {
fun unfollow()
fun shutdown()
}
fun follower() = object : Follower {
private val start = state.locked { registeredShutdowns.size }
private val end = AtomicInteger(start - 1)
override fun unfollow() = end.set(state.locked { registeredShutdowns.size })
override fun shutdown() = end.get().let { end ->
start > end && throw IllegalStateException("You haven't called unfollow.")
state.locked {
registeredShutdowns.subList(start, end).listIterator(end - start).run {
while (hasPrevious()) {
previous().getOrThrow().invoke()
set(doneFuture {}) // Don't break other followers by doing a remove.
}
}
}
}
}
}

View File

@ -1,4 +1,4 @@
package net.corda.testing.driver package net.corda.testing.internal
import net.corda.core.internal.div import net.corda.core.internal.div
import net.corda.core.internal.exists import net.corda.core.internal.exists

View File

@ -1,4 +1,4 @@
package net.corda.testing package net.corda.testing.internal
import net.corda.client.mock.Generator import net.corda.client.mock.Generator
import net.corda.client.rpc.internal.KryoClientSerializationScheme import net.corda.client.rpc.internal.KryoClientSerializationScheme

View File

@ -1,4 +1,4 @@
package net.corda.testing.performance package net.corda.testing.internal.performance
import com.codahale.metrics.Gauge import com.codahale.metrics.Gauge
import com.codahale.metrics.MetricRegistry import com.codahale.metrics.MetricRegistry

View File

@ -1,4 +1,4 @@
package net.corda.testing.performance package net.corda.testing.internal.performance
import com.codahale.metrics.ConsoleReporter import com.codahale.metrics.ConsoleReporter
import com.codahale.metrics.JmxReporter import com.codahale.metrics.JmxReporter

View File

@ -1,4 +1,4 @@
package net.corda.testing.performance package net.corda.testing.internal.performance
import java.time.Duration import java.time.Duration
import java.time.temporal.ChronoUnit import java.time.temporal.ChronoUnit

View File

@ -19,6 +19,7 @@ import net.corda.nodeapi.VerifierApi
import net.corda.nodeapi.config.NodeSSLConfiguration import net.corda.nodeapi.config.NodeSSLConfiguration
import net.corda.nodeapi.config.SSLConfiguration import net.corda.nodeapi.config.SSLConfiguration
import net.corda.testing.driver.* import net.corda.testing.driver.*
import net.corda.testing.internal.ProcessUtilities
import net.corda.testing.node.NotarySpec import net.corda.testing.node.NotarySpec
import org.apache.activemq.artemis.api.core.SimpleString import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.ActiveMQClient import org.apache.activemq.artemis.api.core.client.ActiveMQClient