mirror of
https://github.com/corda/corda.git
synced 2024-12-27 08:22:35 +00:00
Merge pull request #560 from corda/corda/tudor-os-merge
Corda/tudor os merge
This commit is contained in:
Tudor Malene
GitHub
commit
f67c6874f4
@ -47,7 +47,6 @@ dependencies {
|
||||
compile project(':client:jfx')
|
||||
compile project(':node-driver')
|
||||
compile project(':webserver')
|
||||
testCompile project(':verifier')
|
||||
testCompile project(':test-utils')
|
||||
|
||||
compile "org.graphstream:gs-core:1.3"
|
||||
|
||||
@ -13,13 +13,11 @@ package net.corda.docs
|
||||
import net.corda.node.services.config.ConfigHelper
|
||||
import net.corda.node.services.config.NodeConfiguration
|
||||
import net.corda.node.services.config.parseAsNodeConfiguration
|
||||
import net.corda.verifier.Verifier
|
||||
import org.junit.Test
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.Paths
|
||||
import kotlin.reflect.KVisibility
|
||||
import kotlin.reflect.full.declaredMemberProperties
|
||||
import kotlin.reflect.jvm.isAccessible
|
||||
|
||||
class ExampleConfigTest {
|
||||
|
||||
@ -51,14 +49,4 @@ class ExampleConfigTest {
|
||||
).parseAsNodeConfiguration()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `example verifier_conf parses fine`() {
|
||||
readAndCheckConfigurations(
|
||||
"example-verifier.conf"
|
||||
) {
|
||||
val baseDirectory = Paths.get("some-example-base-dir")
|
||||
Verifier.loadConfiguration(baseDirectory, it)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -110,7 +110,7 @@ open class Node(configuration: NodeConfiguration,
|
||||
|
||||
override val log: Logger get() = staticLog
|
||||
override fun makeTransactionVerifierService(): TransactionVerifierService = when (configuration.verifierType) {
|
||||
VerifierType.OutOfProcess -> verifierMessagingClient!!.verifierService
|
||||
VerifierType.OutOfProcess -> throw IllegalArgumentException("OutOfProcess verifier not supported") //verifierMessagingClient!!.verifierService
|
||||
VerifierType.InMemory -> InMemoryTransactionVerifierService(numberOfWorkers = 4)
|
||||
}
|
||||
|
||||
@ -192,7 +192,7 @@ open class Node(configuration: NodeConfiguration,
|
||||
rpcMessagingClient = RPCMessagingClient(configuration.rpcOptions.sslConfig, it.admin, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE, rpcServerConfiguration)
|
||||
}
|
||||
verifierMessagingClient = when (configuration.verifierType) {
|
||||
VerifierType.OutOfProcess -> VerifierMessagingClient(configuration, serverAddress, services.monitoringService.metrics, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE)
|
||||
VerifierType.OutOfProcess -> throw IllegalArgumentException("OutOfProcess verifier not supported") //VerifierMessagingClient(configuration, serverAddress, services.monitoringService.metrics, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE)
|
||||
VerifierType.InMemory -> null
|
||||
}
|
||||
require(info.legalIdentities.size in 1..2) { "Currently nodes must have a primary address and optionally one serviced address" }
|
||||
|
||||
@ -34,7 +34,6 @@ include 'experimental:quasar-hook'
|
||||
include 'experimental:kryo-hook'
|
||||
include 'experimental:intellij-plugin'
|
||||
include 'experimental:flow-hook'
|
||||
include 'verifier'
|
||||
include 'test-common'
|
||||
include 'test-utils'
|
||||
include 'smoke-test-utils'
|
||||
|
||||
@ -1,94 +0,0 @@
|
||||
/*
|
||||
* R3 Proprietary and Confidential
|
||||
*
|
||||
* Copyright (c) 2018 R3 Limited. All rights reserved.
|
||||
*
|
||||
* The intellectual and technical concepts contained herein are proprietary to R3 and its suppliers and are protected by trade secret law.
|
||||
*
|
||||
* Distribution of this file or any portion thereof via any medium without the express permission of R3 is strictly prohibited.
|
||||
*/
|
||||
|
||||
apply plugin: 'kotlin'
|
||||
apply plugin: 'net.corda.plugins.quasar-utils'
|
||||
apply plugin: 'net.corda.plugins.publish-utils'
|
||||
apply plugin: 'com.jfrog.artifactory'
|
||||
|
||||
description 'Corda verifier'
|
||||
|
||||
//noinspection GroovyAssignabilityCheck
|
||||
configurations {
|
||||
integrationTestCompile.extendsFrom testCompile
|
||||
integrationTestRuntime.extendsFrom testRuntime
|
||||
}
|
||||
|
||||
sourceSets {
|
||||
integrationTest {
|
||||
kotlin {
|
||||
compileClasspath += main.output + test.output
|
||||
runtimeClasspath += main.output + test.output
|
||||
srcDir file('src/integration-test/kotlin')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Use manual resource copying of log4j2.xml rather than source sets.
|
||||
// This prevents problems in IntelliJ with regard to duplicate source roots.
|
||||
processResources {
|
||||
from file("$rootDir/config/dev/log4j2.xml")
|
||||
}
|
||||
|
||||
dependencies {
|
||||
compile project(":node-api")
|
||||
|
||||
compile "org.jetbrains.kotlin:kotlin-stdlib-jre8:$kotlin_version"
|
||||
compile "org.jetbrains.kotlin:kotlin-reflect:$kotlin_version"
|
||||
testCompile "org.jetbrains.kotlin:kotlin-test:$kotlin_version"
|
||||
|
||||
// TODO: remove the forced update of commons-collections and beanutils when artemis updates them
|
||||
compile "org.apache.commons:commons-collections4:${commons_collections_version}"
|
||||
compile "commons-beanutils:commons-beanutils:${beanutils_version}"
|
||||
compile "org.apache.activemq:artemis-core-client:${artemis_version}"
|
||||
|
||||
// Log4J: logging framework (with SLF4J bindings)
|
||||
compile "org.apache.logging.log4j:log4j-slf4j-impl:${log4j_version}"
|
||||
compile "org.apache.logging.log4j:log4j-core:${log4j_version}"
|
||||
|
||||
integrationTestCompile project(":node-driver")
|
||||
integrationTestCompile project(":client:mock")
|
||||
|
||||
// Integration test helpers
|
||||
integrationTestCompile "junit:junit:$junit_version"
|
||||
|
||||
integrationTestCompile "org.apache.activemq:artemis-server:${artemis_version}"
|
||||
}
|
||||
|
||||
task standaloneJar(type: Jar) {
|
||||
// Create a fat jar by packing all deps into the output
|
||||
from {
|
||||
configurations.runtime.collect { it.isDirectory() ? it : zipTree(it) }
|
||||
}
|
||||
with jar
|
||||
exclude("META-INF/*.DSA")
|
||||
exclude("META-INF/*.RSA")
|
||||
exclude("META-INF/*.SF")
|
||||
manifest {
|
||||
attributes 'Main-Class': 'net.corda.verifier.Verifier'
|
||||
}
|
||||
archiveName "corda-verifier.jar"
|
||||
}
|
||||
|
||||
task integrationTest(type: Test) {
|
||||
testClassesDirs = sourceSets.integrationTest.output.classesDirs
|
||||
classpath = sourceSets.integrationTest.runtimeClasspath
|
||||
}
|
||||
|
||||
artifacts {
|
||||
publish standaloneJar {
|
||||
classifier ""
|
||||
}
|
||||
}
|
||||
|
||||
publish {
|
||||
disableDefaultJar = true
|
||||
name 'corda-verifier'
|
||||
}
|
||||
@ -1,266 +0,0 @@
|
||||
/*
|
||||
* R3 Proprietary and Confidential
|
||||
*
|
||||
* Copyright (c) 2018 R3 Limited. All rights reserved.
|
||||
*
|
||||
* The intellectual and technical concepts contained herein are proprietary to R3 and its suppliers and are protected by trade secret law.
|
||||
*
|
||||
* Distribution of this file or any portion thereof via any medium without the express permission of R3 is strictly prohibited.
|
||||
*/
|
||||
|
||||
package net.corda.verifier
|
||||
|
||||
import com.nhaarman.mockito_kotlin.any
|
||||
import com.nhaarman.mockito_kotlin.doAnswer
|
||||
import com.nhaarman.mockito_kotlin.whenever
|
||||
import net.corda.client.mock.Generator
|
||||
import net.corda.core.contracts.*
|
||||
import net.corda.core.cordapp.CordappProvider
|
||||
import net.corda.core.crypto.SecureHash
|
||||
import net.corda.core.crypto.entropyToKeyPair
|
||||
import net.corda.core.identity.AbstractParty
|
||||
import net.corda.core.identity.AnonymousParty
|
||||
import net.corda.core.identity.CordaX500Name
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.node.NetworkParameters
|
||||
import net.corda.core.node.ServicesForResolution
|
||||
import net.corda.core.node.services.AttachmentStorage
|
||||
import net.corda.core.node.services.IdentityService
|
||||
import net.corda.core.transactions.LedgerTransaction
|
||||
import net.corda.core.transactions.WireTransaction
|
||||
import net.corda.nodeapi.internal.serialization.GeneratedAttachment
|
||||
import net.corda.testing.common.internal.testNetworkParameters
|
||||
import net.corda.testing.contracts.DummyContract
|
||||
import net.corda.testing.internal.rigorousMock
|
||||
import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
|
||||
import java.math.BigInteger
|
||||
import java.security.PublicKey
|
||||
import java.util.*
|
||||
|
||||
/**
|
||||
* [GeneratedLedger] is a ledger with transactions that always verify.
|
||||
* It provides generator methods, in particular [transactionGenerator] that generates a valid transaction and also
|
||||
* returns the new state of the ledger.
|
||||
*/
|
||||
data class GeneratedLedger(
|
||||
val transactions: List<WireTransaction>,
|
||||
// notary -> outputs. We need to track this because of the unique-notary-on-inputs invariant
|
||||
val availableOutputs: Map<Party, List<StateAndRef<ContractState>>>,
|
||||
val attachments: Set<Attachment>,
|
||||
val identities: Set<Party>
|
||||
) {
|
||||
private val hashTransactionMap: Map<SecureHash, WireTransaction> by lazy { transactions.associateBy(WireTransaction::id) }
|
||||
private val attachmentMap: Map<SecureHash, Attachment> by lazy { attachments.associateBy(Attachment::id) }
|
||||
private val identityMap: Map<PublicKey, Party> by lazy { identities.associateBy(Party::owningKey) }
|
||||
private val contractAttachmentMap: Map<String, ContractAttachment> by lazy {
|
||||
attachments.mapNotNull { it as? ContractAttachment }.flatMap { attch-> attch.allContracts.map { it to attch } }.toMap()
|
||||
}
|
||||
|
||||
private val services = object : ServicesForResolution {
|
||||
override fun loadState(stateRef: StateRef): TransactionState<*> {
|
||||
return hashTransactionMap[stateRef.txhash]?.outputs?.get(stateRef.index) ?: throw TransactionResolutionException(stateRef.txhash)
|
||||
}
|
||||
|
||||
override fun loadStates(stateRefs: Set<StateRef>): Set<StateAndRef<ContractState>> {
|
||||
return stateRefs.groupBy { it.txhash }.flatMap {
|
||||
val outputs = hashTransactionMap[it.key]?.outputs ?: throw TransactionResolutionException(it.key)
|
||||
it.value.map { StateAndRef(outputs[it.index], it) }
|
||||
}.toSet()
|
||||
}
|
||||
override val identityService = rigorousMock<IdentityService>().apply {
|
||||
doAnswer { identityMap[it.arguments[0]] }.whenever(this).partyFromKey(any())
|
||||
}
|
||||
override val attachments = rigorousMock<AttachmentStorage>().apply {
|
||||
doAnswer { attachmentMap[it.arguments[0]] }.whenever(this).openAttachment(any())
|
||||
}
|
||||
override val cordappProvider = rigorousMock<CordappProvider>().apply {
|
||||
doAnswer { contractAttachmentMap[it.arguments[0]]?.id }.whenever(this).getContractAttachmentID(any())
|
||||
}
|
||||
override val networkParameters = testNetworkParameters()
|
||||
}
|
||||
|
||||
companion object {
|
||||
val empty = GeneratedLedger(emptyList(), emptyMap(), emptySet(), emptySet())
|
||||
val contractAttachment = ContractAttachment(GeneratedAttachment(EMPTY_BYTE_ARRAY), DummyContract.PROGRAM_ID)
|
||||
}
|
||||
|
||||
fun resolveWireTransaction(transaction: WireTransaction): LedgerTransaction {
|
||||
return transaction.toLedgerTransaction(services)
|
||||
}
|
||||
|
||||
val attachmentsGenerator: Generator<List<Attachment>> by lazy {
|
||||
// TODO generate contract attachments properly
|
||||
val dummyAttachment = Generator.pure(contractAttachment)
|
||||
val otherAttachments = Generator.replicatePoisson(1.0, pickOneOrMaybeNew(attachments, attachmentGenerator))
|
||||
dummyAttachment.combine(otherAttachments) { dummy, other -> other + dummy }
|
||||
}
|
||||
|
||||
val commandsGenerator: Generator<List<Pair<Command<*>, Party>>> by lazy {
|
||||
Generator.replicatePoisson(4.0, commandGenerator(identities), atLeastOne = true)
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates an issuance(root) transaction.
|
||||
* Invariants: The input list must be empty.
|
||||
*/
|
||||
val issuanceGenerator: Generator<Pair<WireTransaction, GeneratedLedger>> by lazy {
|
||||
val outputsGen = outputsGenerator.flatMap { outputs ->
|
||||
Generator.sequence(
|
||||
outputs.map { output ->
|
||||
pickOneOrMaybeNew(identities, partyGenerator).map { notary ->
|
||||
TransactionState(output, DummyContract.PROGRAM_ID, notary, null, HashAttachmentConstraint(contractAttachment.id))
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
attachmentsGenerator.combine(outputsGen, commandsGenerator) { txAttachments, outputs, commands ->
|
||||
val newTransaction = WireTransaction(
|
||||
emptyList(),
|
||||
txAttachments.map { it.id },
|
||||
outputs,
|
||||
commands.map { it.first },
|
||||
null,
|
||||
null
|
||||
)
|
||||
val newOutputStateAndRefs = outputs.mapIndexed { i, state ->
|
||||
StateAndRef(state, StateRef(newTransaction.id, i))
|
||||
}
|
||||
val newAvailableOutputs = availableOutputs + newOutputStateAndRefs.groupBy { it.state.notary }
|
||||
val newAttachments = attachments + txAttachments
|
||||
val newIdentities = identities + commands.map { it.second } + outputs.map { it.notary }
|
||||
val newLedger = GeneratedLedger(transactions + newTransaction, newAvailableOutputs, newAttachments, newIdentities)
|
||||
Pair(newTransaction, newLedger)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates an exit transaction.
|
||||
* Invariants:
|
||||
* * The output list must be empty
|
||||
*/
|
||||
fun exitTransactionGenerator(inputNotary: Party, inputsToChooseFrom: List<StateAndRef<ContractState>>): Generator<Pair<WireTransaction, GeneratedLedger>> {
|
||||
val inputsGen = Generator.sampleBernoulli(inputsToChooseFrom)
|
||||
return inputsGen.combine(attachmentsGenerator, commandsGenerator) { inputs, txAttachments, commands ->
|
||||
val newTransaction = WireTransaction(
|
||||
inputs.map { it.ref },
|
||||
txAttachments.map { it.id },
|
||||
emptyList(),
|
||||
commands.map { it.first },
|
||||
inputNotary,
|
||||
null
|
||||
)
|
||||
|
||||
val availableOutputsMinusConsumed = HashMap(availableOutputs)
|
||||
if (inputs.size == inputsToChooseFrom.size) {
|
||||
availableOutputsMinusConsumed.remove(inputNotary)
|
||||
} else {
|
||||
availableOutputsMinusConsumed[inputNotary] = inputsToChooseFrom - inputs
|
||||
}
|
||||
val newAvailableOutputs = availableOutputsMinusConsumed
|
||||
val newAttachments = attachments + txAttachments
|
||||
val newIdentities = identities + commands.map { it.second }
|
||||
val newLedger = GeneratedLedger(transactions + newTransaction, newAvailableOutputs, newAttachments, newIdentities)
|
||||
Pair(newTransaction, newLedger)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates a regular non-issue transaction.
|
||||
* Invariants:
|
||||
* * Input and output notaries must be one and the same.
|
||||
* * There must be at least one input and output state.
|
||||
*/
|
||||
fun regularTransactionGenerator(inputNotary: Party, inputsToChooseFrom: List<StateAndRef<ContractState>>): Generator<Pair<WireTransaction, GeneratedLedger>> {
|
||||
val outputsGen = outputsGenerator.map { outputs ->
|
||||
outputs.map { output ->
|
||||
TransactionState(output, DummyContract.PROGRAM_ID, inputNotary, null, HashAttachmentConstraint(contractAttachment.id))
|
||||
}
|
||||
}
|
||||
val inputsGen = Generator.sampleBernoulli(inputsToChooseFrom)
|
||||
return inputsGen.combine(attachmentsGenerator, outputsGen, commandsGenerator) { inputs, txAttachments, outputs, commands ->
|
||||
val newTransaction = WireTransaction(
|
||||
inputs.map { it.ref },
|
||||
txAttachments.map { it.id },
|
||||
outputs,
|
||||
commands.map { it.first },
|
||||
inputNotary,
|
||||
null
|
||||
)
|
||||
val newOutputStateAndRefs = outputs.mapIndexed { i, state ->
|
||||
StateAndRef(state, StateRef(newTransaction.id, i))
|
||||
}
|
||||
val availableOutputsMinusConsumed = HashMap(availableOutputs)
|
||||
if (inputs.size == inputsToChooseFrom.size) {
|
||||
availableOutputsMinusConsumed.remove(inputNotary)
|
||||
} else {
|
||||
availableOutputsMinusConsumed[inputNotary] = inputsToChooseFrom - inputs
|
||||
}
|
||||
val newAvailableOutputs = availableOutputsMinusConsumed + newOutputStateAndRefs.groupBy { it.state.notary }
|
||||
val newAttachments = attachments + txAttachments
|
||||
val newIdentities = identities + commands.map { it.second }
|
||||
val newLedger = GeneratedLedger(transactions + newTransaction, newAvailableOutputs, newAttachments, newIdentities)
|
||||
Pair(newTransaction, newLedger)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates a valid transaction. It may be either an issuance or a regular spend transaction. These have
|
||||
* different invariants on notary fields.
|
||||
*/
|
||||
val transactionGenerator: Generator<Pair<WireTransaction, GeneratedLedger>> by lazy {
|
||||
if (availableOutputs.isEmpty()) {
|
||||
issuanceGenerator
|
||||
} else {
|
||||
Generator.pickOne(availableOutputs.keys.toList()).flatMap { inputNotary ->
|
||||
val inputsToChooseFrom = availableOutputs[inputNotary]!!
|
||||
Generator.frequency(
|
||||
0.3 to issuanceGenerator,
|
||||
0.3 to exitTransactionGenerator(inputNotary, inputsToChooseFrom),
|
||||
0.4 to regularTransactionGenerator(inputNotary, inputsToChooseFrom)
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
data class GeneratedState(
|
||||
val nonce: Long,
|
||||
override val participants: List<AbstractParty>
|
||||
) : ContractState
|
||||
|
||||
class GeneratedCommandData(
|
||||
val nonce: Long
|
||||
) : CommandData
|
||||
|
||||
val keyPairGenerator = Generator.long().map { entropyToKeyPair(BigInteger.valueOf(it)) }
|
||||
val publicKeyGenerator = keyPairGenerator.map { it.public }
|
||||
val stateGenerator: Generator<ContractState> =
|
||||
Generator.replicatePoisson(2.0, publicKeyGenerator).combine(Generator.long()) { participants, nonce ->
|
||||
GeneratedState(nonce, participants.map { AnonymousParty(it) })
|
||||
}
|
||||
|
||||
fun commandGenerator(partiesToPickFrom: Collection<Party>): Generator<Pair<Command<*>, Party>> {
|
||||
return pickOneOrMaybeNew(partiesToPickFrom, partyGenerator).combine(Generator.long()) { signer, nonce ->
|
||||
Pair(
|
||||
Command(GeneratedCommandData(nonce), signer.owningKey),
|
||||
signer
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
val partyGenerator: Generator<Party> = Generator.int().combine(publicKeyGenerator) { n, key ->
|
||||
Party(CordaX500Name(organisation = "Party$n", locality = "London", country = "GB"), key)
|
||||
}
|
||||
|
||||
fun <A> pickOneOrMaybeNew(from: Collection<A>, generator: Generator<A>): Generator<A> {
|
||||
if (from.isEmpty()) {
|
||||
return generator
|
||||
} else {
|
||||
return generator.flatMap {
|
||||
Generator.pickOne(from + it)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val attachmentGenerator: Generator<Attachment> = Generator.bytes(16).map(::GeneratedAttachment)
|
||||
val outputsGenerator = Generator.replicatePoisson(3.0, stateGenerator, atLeastOne = true)
|
||||
@ -1,277 +0,0 @@
|
||||
/*
|
||||
* R3 Proprietary and Confidential
|
||||
*
|
||||
* Copyright (c) 2018 R3 Limited. All rights reserved.
|
||||
*
|
||||
* The intellectual and technical concepts contained herein are proprietary to R3 and its suppliers and are protected by trade secret law.
|
||||
*
|
||||
* Distribution of this file or any portion thereof via any medium without the express permission of R3 is strictly prohibited.
|
||||
*/
|
||||
|
||||
package net.corda.verifier
|
||||
|
||||
import com.typesafe.config.Config
|
||||
import com.typesafe.config.ConfigFactory
|
||||
import net.corda.core.concurrent.CordaFuture
|
||||
import net.corda.core.crypto.random63BitValue
|
||||
import net.corda.core.identity.CordaX500Name
|
||||
import net.corda.core.internal.concurrent.OpenFuture
|
||||
import net.corda.core.internal.concurrent.doneFuture
|
||||
import net.corda.core.internal.concurrent.fork
|
||||
import net.corda.core.internal.concurrent.openFuture
|
||||
import net.corda.core.internal.createDirectories
|
||||
import net.corda.core.internal.div
|
||||
import net.corda.core.node.NetworkParameters
|
||||
import net.corda.core.serialization.internal.nodeSerializationEnv
|
||||
import net.corda.core.transactions.LedgerTransaction
|
||||
import net.corda.core.utilities.NetworkHostAndPort
|
||||
import net.corda.core.utilities.contextLogger
|
||||
import net.corda.node.services.config.configureDevKeyAndTrustStores
|
||||
import net.corda.nodeapi.ArtemisTcpTransport
|
||||
import net.corda.nodeapi.ConnectionDirection
|
||||
import net.corda.nodeapi.VerifierApi
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER
|
||||
import net.corda.nodeapi.internal.config.NodeSSLConfiguration
|
||||
import net.corda.nodeapi.internal.config.SSLConfiguration
|
||||
import net.corda.testing.common.internal.testNetworkParameters
|
||||
import net.corda.testing.driver.*
|
||||
import net.corda.testing.driver.internal.NodeHandleInternal
|
||||
import net.corda.testing.node.NotarySpec
|
||||
import net.corda.testing.node.internal.*
|
||||
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.ClientProducer
|
||||
import org.apache.activemq.artemis.api.core.client.ClientSession
|
||||
import org.apache.activemq.artemis.core.config.Configuration
|
||||
import org.apache.activemq.artemis.core.config.CoreQueueConfiguration
|
||||
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl
|
||||
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptorFactory
|
||||
import org.apache.activemq.artemis.core.security.CheckType
|
||||
import org.apache.activemq.artemis.core.security.Role
|
||||
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl
|
||||
import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.Paths
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.ScheduledExecutorService
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
/**
|
||||
* Behaves the same as [driver] and adds verifier-related functionality.
|
||||
*/
|
||||
fun <A> verifierDriver(
|
||||
defaultParameters: DriverParameters = DriverParameters().copy(
|
||||
notarySpecs = emptyList()
|
||||
),
|
||||
dsl: VerifierDriverDSL.() -> A
|
||||
): A {
|
||||
return genericDriver(
|
||||
defaultParameters = defaultParameters.copy(
|
||||
initialiseSerialization = false
|
||||
),
|
||||
driverDslWrapper = ::VerifierDriverDSL,
|
||||
coerce = { it },
|
||||
dsl = dsl
|
||||
)
|
||||
}
|
||||
|
||||
/** A handle for a verifier */
|
||||
data class VerifierHandle(
|
||||
val process: Process
|
||||
)
|
||||
|
||||
/** A handle for the verification requestor */
|
||||
data class VerificationRequestorHandle(
|
||||
val p2pAddress: NetworkHostAndPort,
|
||||
private val responseAddress: SimpleString,
|
||||
private val session: ClientSession,
|
||||
private val requestProducer: ClientProducer,
|
||||
private val addVerificationFuture: (Long, OpenFuture<Throwable?>) -> Unit,
|
||||
private val executorService: ScheduledExecutorService
|
||||
) {
|
||||
fun verifyTransaction(transaction: LedgerTransaction): CordaFuture<Throwable?> {
|
||||
val message = session.createMessage(false)
|
||||
val verificationId = random63BitValue()
|
||||
val request = VerifierApi.VerificationRequest(verificationId, transaction, responseAddress)
|
||||
request.writeToClientMessage(message)
|
||||
val verificationFuture = openFuture<Throwable?>()
|
||||
addVerificationFuture(verificationId, verificationFuture)
|
||||
requestProducer.send(message)
|
||||
return verificationFuture
|
||||
}
|
||||
|
||||
fun waitUntilNumberOfVerifiers(number: Int) {
|
||||
poll(executorService, "$number verifiers to come online") {
|
||||
if (session.queueQuery(SimpleString(VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME)).consumerCount >= number) {
|
||||
Unit
|
||||
} else {
|
||||
null
|
||||
}
|
||||
}.get()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
data class VerifierDriverDSL(private val driverDSL: DriverDSLImpl) : InternalDriverDSL by driverDSL {
|
||||
private val verifierCount = AtomicInteger(0)
|
||||
|
||||
companion object {
|
||||
private val log = contextLogger()
|
||||
fun createConfiguration(baseDirectory: Path, nodeHostAndPort: NetworkHostAndPort): Config {
|
||||
return ConfigFactory.parseMap(
|
||||
mapOf(
|
||||
"baseDirectory" to baseDirectory.toString(),
|
||||
"nodeHostAndPort" to nodeHostAndPort.toString()
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
fun createVerificationRequestorArtemisConfig(baseDirectory: Path, responseAddress: String, hostAndPort: NetworkHostAndPort, sslConfiguration: SSLConfiguration): Configuration {
|
||||
val connectionDirection = ConnectionDirection.Inbound(acceptorFactoryClassName = NettyAcceptorFactory::class.java.name)
|
||||
return ConfigurationImpl().apply {
|
||||
val artemisDir = "$baseDirectory/artemis"
|
||||
bindingsDirectory = "$artemisDir/bindings"
|
||||
journalDirectory = "$artemisDir/journal"
|
||||
largeMessagesDirectory = "$artemisDir/large-messages"
|
||||
acceptorConfigurations = setOf(ArtemisTcpTransport.tcpTransport(connectionDirection, hostAndPort, sslConfiguration))
|
||||
queueConfigurations = listOf(
|
||||
CoreQueueConfiguration().apply {
|
||||
name = VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME
|
||||
address = VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME
|
||||
isDurable = false
|
||||
},
|
||||
CoreQueueConfiguration().apply {
|
||||
name = responseAddress
|
||||
address = responseAddress
|
||||
isDurable = false
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Starts a lightweight verification requestor that implements the Node's Verifier API */
|
||||
fun startVerificationRequestor(name: CordaX500Name): CordaFuture<VerificationRequestorHandle> {
|
||||
val hostAndPort = driverDSL.portAllocation.nextHostAndPort()
|
||||
return driverDSL.executorService.fork {
|
||||
startVerificationRequestorInternal(name, hostAndPort)
|
||||
}
|
||||
}
|
||||
|
||||
private fun startVerificationRequestorInternal(name: CordaX500Name, hostAndPort: NetworkHostAndPort): VerificationRequestorHandle {
|
||||
val baseDir = driverDSL.driverDirectory / name.organisation
|
||||
val sslConfig = object : NodeSSLConfiguration {
|
||||
override val baseDirectory = baseDir
|
||||
override val keyStorePassword: String get() = "cordacadevpass"
|
||||
override val trustStorePassword: String get() = "trustpass"
|
||||
}
|
||||
sslConfig.configureDevKeyAndTrustStores(name)
|
||||
|
||||
val responseQueueNonce = random63BitValue()
|
||||
val responseAddress = "${VerifierApi.VERIFICATION_RESPONSES_QUEUE_NAME_PREFIX}.$responseQueueNonce"
|
||||
|
||||
val artemisConfig = createVerificationRequestorArtemisConfig(baseDir, responseAddress, hostAndPort, sslConfig)
|
||||
|
||||
val securityManager = object : ActiveMQSecurityManager {
|
||||
// We don't need auth, SSL is good enough
|
||||
override fun validateUser(user: String?, password: String?) = true
|
||||
|
||||
override fun validateUserAndRole(user: String?, password: String?, roles: MutableSet<Role>?, checkType: CheckType?) = true
|
||||
}
|
||||
|
||||
val server = ActiveMQServerImpl(artemisConfig, securityManager)
|
||||
log.info("Starting verification requestor Artemis server with base dir $baseDir")
|
||||
server.start()
|
||||
driverDSL.shutdownManager.registerShutdown(doneFuture {
|
||||
server.stop()
|
||||
})
|
||||
val locator = ActiveMQClient.createServerLocatorWithoutHA().apply {
|
||||
isUseGlobalPools = nodeSerializationEnv != null
|
||||
}
|
||||
val transport = ArtemisTcpTransport.tcpTransport(ConnectionDirection.Outbound(), hostAndPort, sslConfig)
|
||||
val sessionFactory = locator.createSessionFactory(transport)
|
||||
val session = sessionFactory.createSession()
|
||||
driverDSL.shutdownManager.registerShutdown(doneFuture {
|
||||
session.stop()
|
||||
sessionFactory.close()
|
||||
})
|
||||
val producer = session.createProducer(VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME)
|
||||
|
||||
val consumer = session.createConsumer(responseAddress)
|
||||
// We demux the individual txs ourselves to avoid race when a new verifier is added
|
||||
val verificationResponseFutures = ConcurrentHashMap<Long, OpenFuture<Throwable?>>()
|
||||
consumer.setMessageHandler {
|
||||
val result = VerifierApi.VerificationResponse.fromClientMessage(it)
|
||||
val resultFuture = verificationResponseFutures.remove(result.verificationId)
|
||||
log.info("${verificationResponseFutures.size} verifications left")
|
||||
if (resultFuture != null) {
|
||||
resultFuture.set(result.exception)
|
||||
} else {
|
||||
log.warn("Verification requestor $name can't find tx result future with id ${result.verificationId}, possible dupe")
|
||||
}
|
||||
}
|
||||
session.start()
|
||||
return VerificationRequestorHandle(
|
||||
p2pAddress = hostAndPort,
|
||||
responseAddress = SimpleString(responseAddress),
|
||||
session = session,
|
||||
requestProducer = producer,
|
||||
addVerificationFuture = { verificationNonce, future ->
|
||||
verificationResponseFutures.put(verificationNonce, future)
|
||||
},
|
||||
executorService = driverDSL.executorService
|
||||
)
|
||||
}
|
||||
|
||||
/** Starts an out of process verifier connected to [address] */
|
||||
fun startVerifier(address: NetworkHostAndPort): CordaFuture<VerifierHandle> {
|
||||
log.info("Starting verifier connecting to address $address")
|
||||
val id = verifierCount.andIncrement
|
||||
val jdwpPort = if (driverDSL.isDebug) driverDSL.debugPortAllocation.nextPort() else null
|
||||
val verifierName = CordaX500Name(organisation = "Verifier$id", locality = "London", country = "GB")
|
||||
val baseDirectory = (driverDSL.driverDirectory / verifierName.organisation).createDirectories()
|
||||
val config = createConfiguration(baseDirectory, address)
|
||||
val configFilename = "verifier.conf"
|
||||
writeConfig(baseDirectory, configFilename, config)
|
||||
Verifier.loadConfiguration(baseDirectory, baseDirectory / configFilename).configureDevKeyAndTrustStores(verifierName)
|
||||
val process = ProcessUtilities.startJavaProcess<Verifier>(listOf(baseDirectory.toString()), jdwpPort = jdwpPort)
|
||||
driverDSL.shutdownManager.registerProcessShutdown(process)
|
||||
return doneFuture(VerifierHandle(process))
|
||||
}
|
||||
|
||||
/** Starts a verifier connecting to the specified node */
|
||||
fun startVerifier(nodeHandle: NodeHandle): CordaFuture<VerifierHandle> {
|
||||
return startVerifier(nodeHandle.p2pAddress)
|
||||
}
|
||||
|
||||
/** Starts a verifier connecting to the specified requestor */
|
||||
fun startVerifier(verificationRequestorHandle: VerificationRequestorHandle): CordaFuture<VerifierHandle> {
|
||||
return startVerifier(verificationRequestorHandle.p2pAddress)
|
||||
}
|
||||
|
||||
private fun <A> NodeHandleInternal.connectToNode(closure: (ClientSession) -> A): A {
|
||||
val transport = ArtemisTcpTransport.tcpTransport(ConnectionDirection.Outbound(), p2pAddress, configuration)
|
||||
val locator = ActiveMQClient.createServerLocatorWithoutHA(transport)
|
||||
val sessionFactory = locator.createSessionFactory()
|
||||
val session = sessionFactory.createSession(NODE_USER, NODE_USER, false, true, true, locator.isPreAcknowledge, locator.ackBatchSize)
|
||||
return session.use {
|
||||
closure(it)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits until [number] verifiers are listening for verification requests coming from the Node. Check
|
||||
* [VerificationRequestorHandle.waitUntilNumberOfVerifiers] for an equivalent for requestors.
|
||||
*/
|
||||
fun NodeHandleInternal.waitUntilNumberOfVerifiers(number: Int) {
|
||||
connectToNode { session ->
|
||||
poll(driverDSL.executorService, "$number verifiers to come online") {
|
||||
if (session.queueQuery(SimpleString(VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME)).consumerCount >= number) {
|
||||
Unit
|
||||
} else {
|
||||
null
|
||||
}
|
||||
}.get()
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,176 +0,0 @@
|
||||
/*
|
||||
* R3 Proprietary and Confidential
|
||||
*
|
||||
* Copyright (c) 2018 R3 Limited. All rights reserved.
|
||||
*
|
||||
* The intellectual and technical concepts contained herein are proprietary to R3 and its suppliers and are protected by trade secret law.
|
||||
*
|
||||
* Distribution of this file or any portion thereof via any medium without the express permission of R3 is strictly prohibited.
|
||||
*/
|
||||
|
||||
package net.corda.verifier
|
||||
|
||||
import net.corda.core.contracts.TransactionVerificationException
|
||||
import net.corda.core.internal.concurrent.map
|
||||
import net.corda.core.internal.concurrent.transpose
|
||||
import net.corda.core.messaging.startFlow
|
||||
import net.corda.core.transactions.LedgerTransaction
|
||||
import net.corda.core.transactions.WireTransaction
|
||||
import net.corda.core.utilities.OpaqueBytes
|
||||
import net.corda.core.utilities.getOrThrow
|
||||
import net.corda.finance.DOLLARS
|
||||
import net.corda.finance.flows.CashIssueFlow
|
||||
import net.corda.finance.flows.CashPaymentFlow
|
||||
import net.corda.testing.core.ALICE_NAME
|
||||
import net.corda.testing.core.DUMMY_NOTARY_NAME
|
||||
import net.corda.testing.core.SerializationEnvironmentRule
|
||||
import net.corda.testing.driver.DriverParameters
|
||||
import net.corda.testing.driver.VerifierType
|
||||
import net.corda.testing.driver.internal.NodeHandleInternal
|
||||
import net.corda.testing.internal.IntegrationTest
|
||||
import net.corda.testing.internal.IntegrationTestSchemas
|
||||
import net.corda.testing.internal.toDatabaseSchemaName
|
||||
import net.corda.testing.node.NotarySpec
|
||||
import org.junit.ClassRule
|
||||
import org.junit.Ignore
|
||||
import org.junit.Rule
|
||||
import org.junit.Test
|
||||
import java.util.*
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class VerifierTests : IntegrationTest() {
|
||||
companion object {
|
||||
@ClassRule
|
||||
@JvmField
|
||||
val databaseSchemas = IntegrationTestSchemas(ALICE_NAME.toDatabaseSchemaName(), DUMMY_NOTARY_NAME.toDatabaseSchemaName())
|
||||
}
|
||||
@Rule
|
||||
@JvmField
|
||||
val testSerialization = SerializationEnvironmentRule(true)
|
||||
|
||||
private fun generateTransactions(number: Int): List<LedgerTransaction> {
|
||||
var currentLedger = GeneratedLedger.empty
|
||||
val transactions = arrayListOf<WireTransaction>()
|
||||
val random = SplittableRandom()
|
||||
for (i in 0 until number) {
|
||||
val (tx, ledger) = currentLedger.transactionGenerator.generateOrFail(random)
|
||||
transactions.add(tx)
|
||||
currentLedger = ledger
|
||||
}
|
||||
return transactions.map { currentLedger.resolveWireTransaction(it) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `single verifier works with requestor`() {
|
||||
verifierDriver(DriverParameters(extraCordappPackagesToScan = listOf("net.corda.finance.contracts"))) {
|
||||
val aliceFuture = startVerificationRequestor(ALICE_NAME)
|
||||
val transactions = generateTransactions(100)
|
||||
val alice = aliceFuture.get()
|
||||
startVerifier(alice)
|
||||
alice.waitUntilNumberOfVerifiers(1)
|
||||
val results = transactions.map { alice.verifyTransaction(it) }.transpose().get()
|
||||
results.forEach {
|
||||
if (it != null) {
|
||||
throw it
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `single verification fails`() {
|
||||
verifierDriver(DriverParameters(extraCordappPackagesToScan = listOf("net.corda.finance.contracts"))) {
|
||||
val aliceFuture = startVerificationRequestor(ALICE_NAME)
|
||||
// Generate transactions as per usual, but then remove attachments making transaction invalid.
|
||||
val transactions = generateTransactions(1).map { it.copy(attachments = emptyList()) }
|
||||
val alice = aliceFuture.get()
|
||||
startVerifier(alice)
|
||||
alice.waitUntilNumberOfVerifiers(1)
|
||||
val verificationRejection = transactions.map { alice.verifyTransaction(it) }.transpose().get().single()
|
||||
assertTrue { verificationRejection is TransactionVerificationException.MissingAttachmentRejection}
|
||||
assertTrue { verificationRejection!!.message!!.contains("Contract constraints failed, could not find attachment") }
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `multiple verifiers work with requestor`() {
|
||||
verifierDriver {
|
||||
val aliceFuture = startVerificationRequestor(ALICE_NAME)
|
||||
val transactions = generateTransactions(100)
|
||||
val alice = aliceFuture.get()
|
||||
val numberOfVerifiers = 4
|
||||
for (i in 1..numberOfVerifiers) {
|
||||
startVerifier(alice)
|
||||
}
|
||||
alice.waitUntilNumberOfVerifiers(numberOfVerifiers)
|
||||
val results = transactions.map { alice.verifyTransaction(it) }.transpose().get()
|
||||
results.forEach {
|
||||
if (it != null) {
|
||||
throw it
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `verification redistributes on verifier death`() {
|
||||
verifierDriver {
|
||||
val aliceFuture = startVerificationRequestor(ALICE_NAME)
|
||||
val numberOfTransactions = 100
|
||||
val transactions = generateTransactions(numberOfTransactions)
|
||||
val alice = aliceFuture.get()
|
||||
val verifier1 = startVerifier(alice)
|
||||
val verifier2 = startVerifier(alice)
|
||||
startVerifier(alice)
|
||||
alice.waitUntilNumberOfVerifiers(3)
|
||||
val remainingTransactionsCount = AtomicInteger(numberOfTransactions)
|
||||
val futures = transactions.map { transaction ->
|
||||
val future = alice.verifyTransaction(transaction)
|
||||
// Kill verifiers as results are coming in, forcing artemis to redistribute.
|
||||
future.map {
|
||||
val remaining = remainingTransactionsCount.decrementAndGet()
|
||||
when (remaining) {
|
||||
33 -> verifier1.get().process.destroy()
|
||||
66 -> verifier2.get().process.destroy()
|
||||
}
|
||||
it
|
||||
}
|
||||
}
|
||||
futures.transpose().get()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `verification request waits until verifier comes online`() {
|
||||
verifierDriver {
|
||||
val aliceFuture = startVerificationRequestor(ALICE_NAME)
|
||||
val transactions = generateTransactions(100)
|
||||
val alice = aliceFuture.get()
|
||||
val futures = transactions.map { alice.verifyTransaction(it) }
|
||||
startVerifier(alice)
|
||||
futures.transpose().get()
|
||||
}
|
||||
}
|
||||
|
||||
@Ignore("CORDA-1022")
|
||||
@Test
|
||||
fun `single verifier works with a node`() {
|
||||
verifierDriver(DriverParameters(
|
||||
extraCordappPackagesToScan = listOf("net.corda.finance.contracts", "net.corda.finance.schemas"),
|
||||
notarySpecs = listOf(NotarySpec(DUMMY_NOTARY_NAME, verifierType = VerifierType.OutOfProcess))
|
||||
)) {
|
||||
val aliceNode = startNode(providedName = ALICE_NAME).getOrThrow()
|
||||
val notaryNode = defaultNotaryNode.getOrThrow() as NodeHandleInternal
|
||||
val alice = aliceNode.rpc.wellKnownPartyFromX500Name(ALICE_NAME)!!
|
||||
startVerifier(notaryNode)
|
||||
aliceNode.rpc.startFlow(::CashIssueFlow, 10.DOLLARS, OpaqueBytes.of(0), defaultNotaryIdentity).returnValue.get()
|
||||
notaryNode.waitUntilNumberOfVerifiers(1)
|
||||
for (i in 1..10) {
|
||||
val cashFlowResult = aliceNode.rpc.startFlow(::CashPaymentFlow, 10.DOLLARS, alice).returnValue.get()
|
||||
assertNotNull(cashFlowResult)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,126 +0,0 @@
|
||||
/*
|
||||
* R3 Proprietary and Confidential
|
||||
*
|
||||
* Copyright (c) 2018 R3 Limited. All rights reserved.
|
||||
*
|
||||
* The intellectual and technical concepts contained herein are proprietary to R3 and its suppliers and are protected by trade secret law.
|
||||
*
|
||||
* Distribution of this file or any portion thereof via any medium without the express permission of R3 is strictly prohibited.
|
||||
*/
|
||||
|
||||
package net.corda.verifier
|
||||
|
||||
import com.typesafe.config.Config
|
||||
import com.typesafe.config.ConfigFactory
|
||||
import com.typesafe.config.ConfigParseOptions
|
||||
import net.corda.core.internal.div
|
||||
import net.corda.core.serialization.SerializationContext
|
||||
import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
|
||||
import net.corda.core.serialization.internal.SerializationEnvironmentImpl
|
||||
import net.corda.core.serialization.internal.nodeSerializationEnv
|
||||
import net.corda.core.utilities.*
|
||||
import net.corda.nodeapi.ArtemisTcpTransport.Companion.tcpTransport
|
||||
import net.corda.nodeapi.ConnectionDirection
|
||||
import net.corda.nodeapi.VerifierApi
|
||||
import net.corda.nodeapi.VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME
|
||||
import net.corda.nodeapi.internal.config.NodeSSLConfiguration
|
||||
import net.corda.nodeapi.internal.config.getValue
|
||||
import net.corda.nodeapi.internal.addShutdownHook
|
||||
import net.corda.nodeapi.internal.serialization.*
|
||||
import net.corda.nodeapi.internal.serialization.amqp.AbstractAMQPSerializationScheme
|
||||
import net.corda.nodeapi.internal.serialization.amqp.amqpMagic
|
||||
import net.corda.nodeapi.internal.serialization.amqp.SerializerFactory
|
||||
import net.corda.nodeapi.internal.serialization.kryo.AbstractKryoSerializationScheme
|
||||
import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
|
||||
import org.apache.activemq.artemis.api.core.client.ActiveMQClient
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.Paths
|
||||
|
||||
data class VerifierConfiguration(
|
||||
override val baseDirectory: Path,
|
||||
val config: Config // NB: This property is being used via reflection.
|
||||
) : NodeSSLConfiguration {
|
||||
val nodeHostAndPort: NetworkHostAndPort by config
|
||||
override val keyStorePassword: String by config
|
||||
override val trustStorePassword: String by config
|
||||
}
|
||||
|
||||
class Verifier {
|
||||
companion object {
|
||||
private val log = contextLogger()
|
||||
fun loadConfiguration(baseDirectory: Path, configPath: Path): VerifierConfiguration {
|
||||
val defaultConfig = ConfigFactory.parseResources("verifier-reference.conf", ConfigParseOptions.defaults().setAllowMissing(false))
|
||||
val customConfig = ConfigFactory.parseFile(configPath.toFile(), ConfigParseOptions.defaults().setAllowMissing(false))
|
||||
val resolvedConfig = customConfig.withFallback(defaultConfig).resolve()
|
||||
return VerifierConfiguration(baseDirectory, resolvedConfig)
|
||||
}
|
||||
|
||||
@JvmStatic
|
||||
fun main(args: Array<String>) {
|
||||
require(args.isNotEmpty()) { "Usage: <binary> BASE_DIR_CONTAINING_VERIFIER_CONF" }
|
||||
val baseDirectory = Paths.get(args[0])
|
||||
val verifierConfig = loadConfiguration(baseDirectory, baseDirectory / "verifier.conf")
|
||||
initialiseSerialization()
|
||||
val locator = ActiveMQClient.createServerLocatorWithHA(
|
||||
tcpTransport(ConnectionDirection.Outbound(), verifierConfig.nodeHostAndPort, verifierConfig)
|
||||
)
|
||||
val sessionFactory = locator.createSessionFactory()
|
||||
val session = sessionFactory.createSession(
|
||||
VerifierApi.VERIFIER_USERNAME, VerifierApi.VERIFIER_USERNAME, false, true, true, locator.isPreAcknowledge, locator.ackBatchSize
|
||||
)
|
||||
addShutdownHook {
|
||||
log.info("Shutting down")
|
||||
session.close()
|
||||
sessionFactory.close()
|
||||
}
|
||||
val consumer = session.createConsumer(VERIFICATION_REQUESTS_QUEUE_NAME)
|
||||
val replyProducer = session.createProducer()
|
||||
consumer.setMessageHandler {
|
||||
val (request, context) = VerifierApi.VerificationRequest.fromClientMessage(it)
|
||||
log.debug { "Received verification request with id ${request.verificationId}" }
|
||||
val error = try {
|
||||
request.transaction.verify()
|
||||
null
|
||||
} catch (t: Throwable) {
|
||||
log.debug("Verification returned with error:", t)
|
||||
t
|
||||
}
|
||||
val reply = session.createMessage(false)
|
||||
val response = VerifierApi.VerificationResponse(request.verificationId, error)
|
||||
response.writeToClientMessage(reply, context)
|
||||
replyProducer.send(request.responseAddress, reply)
|
||||
it.acknowledge()
|
||||
}
|
||||
session.start()
|
||||
log.info("Verifier started")
|
||||
Thread.sleep(Long.MAX_VALUE)
|
||||
}
|
||||
|
||||
private fun initialiseSerialization() {
|
||||
nodeSerializationEnv = SerializationEnvironmentImpl(
|
||||
SerializationFactoryImpl().apply {
|
||||
registerScheme(KryoVerifierSerializationScheme)
|
||||
registerScheme(AMQPVerifierSerializationScheme)
|
||||
},
|
||||
AMQP_P2P_CONTEXT)
|
||||
}
|
||||
}
|
||||
|
||||
private object KryoVerifierSerializationScheme : AbstractKryoSerializationScheme() {
|
||||
override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
|
||||
return magic == kryoMagic && target == SerializationContext.UseCase.P2P
|
||||
}
|
||||
|
||||
override fun rpcClientKryoPool(context: SerializationContext) = throw UnsupportedOperationException()
|
||||
override fun rpcServerKryoPool(context: SerializationContext) = throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
private object AMQPVerifierSerializationScheme : AbstractAMQPSerializationScheme(emptyList()) {
|
||||
override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
|
||||
return magic == amqpMagic && target == SerializationContext.UseCase.P2P
|
||||
}
|
||||
|
||||
override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory = throw UnsupportedOperationException()
|
||||
override fun rpcServerSerializerFactory(context: SerializationContext): SerializerFactory = throw UnsupportedOperationException()
|
||||
}
|
||||
}
|
||||
@ -1,3 +0,0 @@
|
||||
# nodeHostAndPort = "localhost:12345"
|
||||
keyStorePassword = "cordacadevpass"
|
||||
trustStorePassword = "trustpass"
|
||||
Loading…
Reference in New Issue
Block a user