ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2024-12-19 04:57:58 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge pull request #720 from corda/shams-flow-registration-annotation

Browse Source 
Introducing InitiatedBy annotation to be used on initiated flows to s…
This commit is contained in:
Shams Asari 2017-05-31 11:17:23 +01:00 committed by GitHub
commit 37e183652e
66 changed files with 892 additions and 760 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
core/src/main/kotlin/net/corda/core/Utils.kt
View File

@ -147,6 +147,12 @@ operator fun String.div(other: String) = Paths.get(this) / other
fun Path.createDirectory(vararg attrs: FileAttribute<*>): Path = Files.createDirectory(this, *attrs)
fun Path.createDirectories(vararg attrs: FileAttribute<*>): Path = Files.createDirectories(this, *attrs)
fun Path.exists(vararg options: LinkOption): Boolean = Files.exists(this, *options)
fun Path.copyToDirectory(targetDir: Path, vararg options: CopyOption): Path {
require(targetDir.isDirectory()) { "$targetDir is not a directory" }
val targetFile = targetDir.resolve(fileName)
Files.copy(this, targetFile, *options)
return targetFile
}
fun Path.moveTo(target: Path, vararg options: CopyOption): Path = Files.move(this, target, *options)
fun Path.isRegularFile(vararg options: LinkOption): Boolean = Files.isRegularFile(this, *options)
fun Path.isDirectory(vararg options: LinkOption): Boolean = Files.isDirectory(this, *options)

16
core/src/main/kotlin/net/corda/core/flows/InitiatedBy.kt Normal file
View File

@ -0,0 +1,16 @@
package net.corda.core.flows
import kotlin.annotation.AnnotationTarget.CLASS
import kotlin.reflect.KClass
/**
* This annotation is required by any [FlowLogic] that is designed to be initiated by a counterparty flow. The flow that
* does the initiating is specified by the [value] property and itself must be annotated with [InitiatingFlow].
*
* The node on startup scans for [FlowLogic]s which are annotated with this and automatically registers the initiating
* to initiated flow mapping.
*
* @see InitiatingFlow
*/
@Target(CLASS)
annotation class InitiatedBy(val value: KClass<out FlowLogic<*>>)

8
core/src/main/kotlin/net/corda/core/flows/InitiatingFlow.kt
View File

@ -5,8 +5,7 @@ import kotlin.annotation.AnnotationTarget.CLASS
/**
* This annotation is required by any [FlowLogic] which has been designated to initiate communication with a counterparty
* and request they start their side of the flow communication. To ensure that this is correctly applied
* [net.corda.core.node.PluginServiceHub.registerServiceFlow] checks the initiating flow class has this annotation.
* and request they start their side of the flow communication.
*
* There is also an optional [version] property, which defaults to 1, to specify the version of the flow protocol. This
* integer value should be incremented whenever there is a release of this flow which has changes that are not backwards
@ -19,6 +18,11 @@ import kotlin.annotation.AnnotationTarget.CLASS
*
* The flow version number is similar in concept to Corda's platform version but they are not the same. A flow's version
* number can change independently of the platform version.
*
* If you are customising an existing initiating flow by sub-classing it then there's no need to specify this annotation
* again. In fact doing so is an error and checks are made to make sure this doesn't occur.
*
* @see InitiatedBy
*/
// TODO Add support for multiple versions once CorDapps are loaded in separate class loaders
@Target(CLASS)

2
core/src/main/kotlin/net/corda/core/flows/StartableByRPC.kt
View File

@ -1,6 +1,5 @@
package net.corda.core.flows
import java.lang.annotation.Inherited
import kotlin.annotation.AnnotationTarget.CLASS
/**
@ -9,7 +8,6 @@ import kotlin.annotation.AnnotationTarget.CLASS
* flow will not be allowed to start and an exception will be thrown.
*/
@Target(CLASS)
@Inherited
@MustBeDocumented
// TODO Consider a different name, something along the lines of SchedulableFlow
annotation class StartableByRPC

36
core/src/main/kotlin/net/corda/core/flows/TxKeyFlow.kt
View File

@ -15,7 +15,7 @@ import java.security.cert.X509Certificate
* This is intended for use as a subflow of another flow.
*/
object TxKeyFlow {
abstract class AbstractIdentityFlow(val otherSide: Party, val revocationEnabled: Boolean): FlowLogic<Map<Party, AnonymousIdentity>>() {
abstract class AbstractIdentityFlow<out T>(val otherSide: Party): FlowLogic<T>() {
fun validateIdentity(untrustedIdentity: Pair<X509CertificateHolder, CertPath>): AnonymousIdentity {
val (wellKnownCert, certPath) = untrustedIdentity
val theirCert = certPath.certificates.last()
@ -26,20 +26,21 @@ object TxKeyFlow {
val anonymousParty = AnonymousParty(theirCert.publicKey)
serviceHub.identityService.registerPath(wellKnownCert, anonymousParty, certPath)
AnonymousIdentity(certPath, X509CertificateHolder(theirCert.encoded), anonymousParty)
} else
throw IllegalStateException("Expected certificate subject to be ${otherSide.name} but found ${certName}")
} else
} else {
throw IllegalStateException("Expected certificate subject to be ${otherSide.name} but found $certName")
}
} else {
throw IllegalStateException("Expected an X.509 certificate but received ${theirCert.javaClass.name}")
}
}
}
@StartableByRPC
@InitiatingFlow
class Requester(otherSide: Party,
revocationEnabled: Boolean,
override val progressTracker: ProgressTracker) : AbstractIdentityFlow(otherSide, revocationEnabled) {
constructor(otherSide: Party,
revocationEnabled: Boolean) : this(otherSide, revocationEnabled, tracker())
val revocationEnabled: Boolean,
override val progressTracker: ProgressTracker) : AbstractIdentityFlow<Map<Party, AnonymousIdentity>>(otherSide) {
constructor(otherSide: Party, revocationEnabled: Boolean) : this(otherSide, revocationEnabled, tracker())
companion object {
object AWAITING_KEY : ProgressTracker.Step("Awaiting key")
@ -62,26 +63,21 @@ object TxKeyFlow {
* Flow which waits for a key request from a counterparty, generates a new key and then returns it to the
* counterparty and as the result from the flow.
*/
class Provider(otherSide: Party,
revocationEnabled: Boolean,
override val progressTracker: ProgressTracker) : AbstractIdentityFlow(otherSide,revocationEnabled) {
constructor(otherSide: Party,
revocationEnabled: Boolean = false) : this(otherSide, revocationEnabled, tracker())
@InitiatedBy(Requester::class)
class Provider(otherSide: Party) : AbstractIdentityFlow<Unit>(otherSide) {
companion object {
object SENDING_KEY : ProgressTracker.Step("Sending key")
fun tracker() = ProgressTracker(SENDING_KEY)
}
override val progressTracker: ProgressTracker = ProgressTracker(SENDING_KEY)
@Suspendable
override fun call(): Map<Party, AnonymousIdentity> {
override fun call() {
val revocationEnabled = false
progressTracker.currentStep = SENDING_KEY
val myIdentityFragment = serviceHub.keyManagementService.freshKeyAndCert(serviceHub.myInfo.legalIdentity, revocationEnabled)
send(otherSide, myIdentityFragment)
val theirIdentity = receive<Pair<X509CertificateHolder, CertPath>>(otherSide).unwrap { validateIdentity(it) }
return mapOf(Pair(otherSide, AnonymousIdentity(myIdentityFragment)),
Pair(serviceHub.myInfo.legalIdentity, theirIdentity))
receive<Pair<X509CertificateHolder, CertPath>>(otherSide).unwrap { validateIdentity(it) }
}
}

3
core/src/main/kotlin/net/corda/core/node/CordaPluginRegistry.kt
View File

@ -33,6 +33,9 @@ abstract class CordaPluginRegistry {
* The [PluginServiceHub] will be fully constructed before the plugin service is created and will
* allow access to the Flow factory and Flow initiation entry points there.
*/
@Suppress("unused")
@Deprecated("This is no longer used. If you need to create your own service, such as an oracle, then use the " +
"@CordaService annotation. For flow registrations use @InitiatedBy.", level = DeprecationLevel.ERROR)
open val servicePlugins: List<Function<PluginServiceHub, out Any>> get() = emptyList()
/**

21
core/src/main/kotlin/net/corda/core/node/PluginServiceHub.kt
View File

@ -7,22 +7,7 @@ import net.corda.core.identity.Party
* A service hub to be used by the [CordaPluginRegistry]
*/
interface PluginServiceHub : ServiceHub {
/**
* Register the service flow factory to use when an initiating party attempts to communicate with us. The registration
* is done against the [Class] object of the client flow to the service flow. What this means is if a counterparty
* starts a [FlowLogic] represented by [initiatingFlowClass] and starts communication with us, we will execute the service
* flow produced by [serviceFlowFactory]. This service flow has respond correctly to the sends and receives the client
* does.
* @param initiatingFlowClass [Class] of the client flow involved in this client-server communication.
* @param serviceFlowFactory Lambda which produces a new service flow for each new client flow communication. The
* [Party] parameter of the factory is the client's identity.
* @throws IllegalArgumentException If [initiatingFlowClass] is not annotated with [net.corda.core.flows.InitiatingFlow].
*/
fun registerServiceFlow(initiatingFlowClass: Class<out FlowLogic<*>>, serviceFlowFactory: (Party) -> FlowLogic<*>)
@Suppress("UNCHECKED_CAST")
@Deprecated("This is scheduled to be removed in a future release", ReplaceWith("registerServiceFlow"))
fun registerFlowInitiator(markerClass: Class<*>, flowFactory: (Party) -> FlowLogic<*>) {
registerServiceFlow(markerClass as Class<out FlowLogic<*>>, flowFactory)
}
@Deprecated("This is no longer used. Instead annotate the flows produced by your factory with @InitiatedBy and have " +
"them point to the initiating flow class.", level = DeprecationLevel.ERROR)
fun registerFlowInitiator(initiatingFlowClass: Class<out FlowLogic<*>>, serviceFlowFactory: (Party) -> FlowLogic<*>) = Unit
}

8
core/src/main/kotlin/net/corda/core/node/ServiceHub.kt
View File

@ -3,6 +3,7 @@ package net.corda.core.node
import net.corda.core.contracts.*
import net.corda.core.crypto.DigitalSignature
import net.corda.core.node.services.*
import net.corda.core.serialization.SerializeAsToken
import net.corda.core.transactions.SignedTransaction
import net.corda.core.transactions.TransactionBuilder
import java.security.PublicKey
@ -44,6 +45,13 @@ interface ServiceHub : ServicesForResolution {
val clock: Clock
val myInfo: NodeInfo
/**
* Return the singleton instance of the given Corda service type. This is a class that is annotated with
* [CordaService] and will have automatically been registered by the node.
* @throws IllegalArgumentException If [type] is not annotated with [CordaService] or if the instance is not found.
*/
fun <T : SerializeAsToken> cordaService(type: Class<T>): T
/**
* Given a [SignedTransaction], writes it to the local storage for validated transactions and then
* sends them to the vault for further processing. Expects to be run within a database transaction.

20
core/src/main/kotlin/net/corda/core/node/services/CordaService.kt Normal file
View File

@ -0,0 +1,20 @@
package net.corda.core.node.services
import kotlin.annotation.AnnotationTarget.CLASS
/**
* Annotate any class that needs to be a long-lived service within the node, such as an oracle, with this annotation.
* Such a class needs to have a constructor with a single parameter of type [net.corda.core.node.PluginServiceHub]. This
* construtor will be invoked during node start to initialise the service. The service hub provided can be used to get
* information about the node that may be necessary for the service. Corda services are created as singletons within
* the node and are available to flows via [net.corda.core.node.ServiceHub.cordaService].
*
* The service class has to implement [net.corda.core.serialization.SerializeAsToken] to ensure correct usage within flows.
* (If possible extend [net.corda.core.serialization.SingletonSerializeAsToken] instead as it removes the boilerplate.)
*/
// TODO Handle the singleton serialisation of Corda services automatically, removing the need to implement SerializeAsToken
// TODO Currently all nodes which load the plugin will attempt to load the service even if it's not revelant to them. The
// underlying problem is that the entire CorDapp jar is used as a dependency, when in fact it's just the client-facing
// bit of the CorDapp that should be depended on (e.g. the initiating flows).
@Target(CLASS)
annotation class CordaService

8
core/src/main/kotlin/net/corda/core/utilities/ProcessUtilities.kt
View File

@ -2,21 +2,24 @@ package net.corda.core.utilities
import java.nio.file.Path
// TODO This doesn't belong in core and can be moved into node
object ProcessUtilities {
inline fun <reified C : Any> startJavaProcess(
arguments: List<String>,
classpath: String = defaultClassPath,
jdwpPort: Int? = null,
extraJvmArguments: List<String> = emptyList(),
inheritIO: Boolean = true,
errorLogPath: Path? = null,
workingDirectory: Path? = null
): Process {
return startJavaProcess(C::class.java.name, arguments, jdwpPort, extraJvmArguments, inheritIO, errorLogPath, workingDirectory)
return startJavaProcess(C::class.java.name, arguments, classpath, jdwpPort, extraJvmArguments, inheritIO, errorLogPath, workingDirectory)
}
fun startJavaProcess(
className: String,
arguments: List<String>,
classpath: String = defaultClassPath,
jdwpPort: Int? = null,
extraJvmArguments: List<String> = emptyList(),
inheritIO: Boolean = true,
@ -24,7 +27,6 @@ object ProcessUtilities {
workingDirectory: Path? = null
): Process {
val separator = System.getProperty("file.separator")
val classpath = System.getProperty("java.class.path")
val javaPath = System.getProperty("java.home") + separator + "bin" + separator + "java"
val debugPortArgument = if (jdwpPort != null) {
listOf("-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=$jdwpPort")
@ -44,4 +46,6 @@ object ProcessUtilities {
if (workingDirectory != null) directory(workingDirectory.toFile())
}.start()
}
val defaultClassPath: String get() = System.getProperty("java.class.path")
}

24
core/src/test/java/net/corda/core/flows/FlowsInJavaTest.java
View File

@ -1,13 +1,15 @@
package net.corda.core.flows;
import co.paralleluniverse.fibers.*;
import co.paralleluniverse.fibers.Suspendable;
import net.corda.core.identity.Party;
import net.corda.testing.node.*;
import org.junit.*;
import net.corda.testing.node.MockNetwork;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.util.concurrent.*;
import java.util.concurrent.Future;
import static org.assertj.core.api.AssertionsForClassTypes.*;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
public class FlowsInJavaTest {
@ -30,13 +32,12 @@ public class FlowsInJavaTest {
@Test
public void suspendableActionInsideUnwrap() throws Exception {
node2.getServices().registerServiceFlow(SendInUnwrapFlow.class, (otherParty) -> new OtherFlow(otherParty, "Hello"));
node2.registerInitiatedFlow(SendHelloAndThenReceive.class);
Future<String> result = node1.getServices().startFlow(new SendInUnwrapFlow(node2.getInfo().getLegalIdentity())).getResultFuture();
net.runNetwork();
assertThat(result.get()).isEqualTo("Hello");
}
@SuppressWarnings("unused")
@InitiatingFlow
private static class SendInUnwrapFlow extends FlowLogic<String> {
private final Party otherParty;
@ -55,19 +56,18 @@ public class FlowsInJavaTest {
}
}
private static class OtherFlow extends FlowLogic<String> {
@InitiatedBy(SendInUnwrapFlow.class)
private static class SendHelloAndThenReceive extends FlowLogic<String> {
private final Party otherParty;
private final String payload;
private OtherFlow(Party otherParty, String payload) {
private SendHelloAndThenReceive(Party otherParty) {
this.otherParty = otherParty;
this.payload = payload;
}
@Suspendable
@Override
public String call() throws FlowException {
return sendAndReceive(String.class, otherParty, payload).unwrap(data -> data);
return sendAndReceive(String.class, otherParty, "Hello").unwrap(data -> data);
}
}

26
core/src/test/kotlin/net/corda/core/flows/CollectSignaturesFlowTests.kt
View File

@ -7,7 +7,6 @@ import net.corda.core.contracts.TransactionType
import net.corda.core.contracts.requireThat
import net.corda.core.getOrThrow
import net.corda.core.identity.Party
import net.corda.core.node.PluginServiceHub
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.unwrap
import net.corda.flows.CollectSignaturesFlow
@ -18,7 +17,7 @@ import net.corda.testing.node.MockNetwork
import org.junit.After
import org.junit.Before
import org.junit.Test
import java.util.concurrent.ExecutionException
import kotlin.reflect.KClass
import kotlin.test.assertFailsWith
class CollectSignaturesFlowTests {
@ -37,9 +36,6 @@ class CollectSignaturesFlowTests {
c = nodes.partyNodes[2]
notary = nodes.notaryNode.info.notaryIdentity
mockNet.runNetwork()
CollectSigsTestCorDapp.registerFlows(a.services)
CollectSigsTestCorDapp.registerFlows(b.services)
CollectSigsTestCorDapp.registerFlows(c.services)
}
@After
@ -47,11 +43,9 @@ class CollectSignaturesFlowTests {
mockNet.stopNodes()
}
object CollectSigsTestCorDapp {
// Would normally be called by custom service init in a CorDapp.
fun registerFlows(pluginHub: PluginServiceHub) {
pluginHub.registerFlowInitiator(TestFlow.Initiator::class.java) { TestFlow.Responder(it) }
pluginHub.registerFlowInitiator(TestFlowTwo.Initiator::class.java) { TestFlowTwo.Responder(it) }
private fun registerFlowOnAllNodes(flowClass: KClass<out FlowLogic<*>>) {
listOf(a, b, c).forEach {
it.registerInitiatedFlow(flowClass.java)
}
}
@ -82,6 +76,7 @@ class CollectSignaturesFlowTests {
}
}
@InitiatedBy(TestFlow.Initiator::class)
class Responder(val otherParty: Party) : FlowLogic<SignedTransaction>() {
@Suspendable
override fun call(): SignedTransaction {
@ -104,7 +99,7 @@ class CollectSignaturesFlowTests {
// receiving off the wire.
object TestFlowTwo {
@InitiatingFlow
class Initiator(val state: DummyContract.MultiOwnerState, val otherParty: Party) : FlowLogic<SignedTransaction>() {
class Initiator(val state: DummyContract.MultiOwnerState) : FlowLogic<SignedTransaction>() {
@Suspendable
override fun call(): SignedTransaction {
val notary = serviceHub.networkMapCache.notaryNodes.single().notaryIdentity
@ -118,6 +113,7 @@ class CollectSignaturesFlowTests {
}
}
@InitiatedBy(TestFlowTwo.Initiator::class)
class Responder(val otherParty: Party) : FlowLogic<SignedTransaction>() {
@Suspendable override fun call(): SignedTransaction {
val flow = object : SignTransactionFlow(otherParty) {
@ -137,13 +133,13 @@ class CollectSignaturesFlowTests {
}
}
@Test
fun `successfully collects two signatures`() {
registerFlowOnAllNodes(TestFlowTwo.Responder::class)
val magicNumber = 1337
val parties = listOf(a.info.legalIdentity, b.info.legalIdentity, c.info.legalIdentity)
val state = DummyContract.MultiOwnerState(magicNumber, parties)
val flow = a.services.startFlow(TestFlowTwo.Initiator(state, b.info.legalIdentity))
val flow = a.services.startFlow(TestFlowTwo.Initiator(state))
mockNet.runNetwork()
val result = flow.resultFuture.getOrThrow()
result.verifySignatures()
@ -169,8 +165,8 @@ class CollectSignaturesFlowTests {
val ptx = onePartyDummyContract.signWith(MINI_CORP_KEY).toSignedTransaction(false)
val flow = a.services.startFlow(CollectSignaturesFlow(ptx))
mockNet.runNetwork()
assertFailsWith<ExecutionException>("The Initiator of CollectSignaturesFlow must have signed the transaction.") {
flow.resultFuture.get()
assertFailsWith<IllegalArgumentException>("The Initiator of CollectSignaturesFlow must have signed the transaction.") {
flow.resultFuture.getOrThrow()
}
}

2
core/src/test/kotlin/net/corda/core/flows/TxKeyFlowTests.kt
View File

@ -38,7 +38,7 @@ class TxKeyFlowTests {
bobNode.services.identityService.registerIdentity(notaryNode.info.legalIdentity)
// Run the flows
bobNode.registerServiceFlow(TxKeyFlow.Requester::class) { TxKeyFlow.Provider(it) }
bobNode.registerInitiatedFlow(TxKeyFlow.Provider::class.java)
val requesterFlow = aliceNode.services.startFlow(TxKeyFlow.Requester(bob, revocationEnabled))
// Get the results

8
core/src/test/kotlin/net/corda/core/serialization/AttachmentSerializationTest.kt
View File

@ -12,10 +12,12 @@ import net.corda.core.messaging.SingleMessageRecipient
import net.corda.core.node.services.ServiceInfo
import net.corda.core.utilities.unwrap
import net.corda.flows.FetchAttachmentsFlow
import net.corda.node.internal.InitiatedFlowFactory
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.network.NetworkMapService
import net.corda.node.services.persistence.NodeAttachmentService
import net.corda.node.services.persistence.schemas.AttachmentEntity
import net.corda.node.services.statemachine.SessionInit
import net.corda.node.utilities.transaction
import net.corda.testing.node.MockNetwork
import org.junit.After
@ -136,7 +138,11 @@ class AttachmentSerializationTest {
}
private fun launchFlow(clientLogic: ClientLogic, rounds: Int) {
server.services.registerServiceFlow(clientLogic.javaClass, ::ServerLogic)
server.registerFlowFactory(ClientLogic::class.java, object : InitiatedFlowFactory<ServerLogic> {
override fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): ServerLogic {
return ServerLogic(otherParty)
}
}, ServerLogic::class.java, track = false)
client.services.startFlow(clientLogic)
network.runNetwork(rounds)
}

51
docs/source/changelog.rst
View File

@ -7,30 +7,26 @@ from the previous milestone release.
UNRELEASED
----------
* API changes:
* ``Timestamp`` used for validation/notarization time-range has been renamed to ``TimeWindow``.
There are now 4 factory methods ``TimeWindow.fromOnly(fromTime: Instant)``,
``TimeWindow.untilOnly(untilTime: Instant)``, ``TimeWindow.between(fromTime: Instant, untilTime: Instant)`` and
``TimeWindow.withTolerance(time: Instant, tolerance: Duration)``.
Previous constructors ``TimeWindow(fromTime: Instant, untilTime: Instant)`` and
``TimeWindow(time: Instant, tolerance: Duration)`` have been removed.
* Quite a few changes have been made to the flow API which should make things simpler when writing CorDapps:
* ``CordaPluginRegistry.requiredFlows`` is no longer needed. Instead annotate any flows you wish to start via RPC with
``@StartableByRPC`` and any scheduled flows with ``@SchedulableFlow``.
* Flows which initiate flows in their counterparties (an example of which is the ``NotaryFlow.Client``) are now
required to be annotated with ``@InitiatingFlow``.
* ``CordaPluginRegistry.servicePlugins`` is also no longer used, along with ``PluginServiceHub.registerFlowInitiator``.
Instead annotate your initiated flows with ``@InitiatedBy``. This annotation takes a single parameter which is the
initiating flow. This initiating flow further has to be annotated with ``@InitiatingFlow``. For any services you
may have, such as oracles, annotate them with ``@CordaService``.
* ``PluginServiceHub.registerFlowInitiator`` has been deprecated and replaced by ``registerServiceFlow`` with the
marker Class restricted to ``FlowLogic``. In line with the introduction of ``InitiatingFlow``, it throws an
``IllegalArgumentException`` if the initiating flow class is not annotated with it.
* Related to ``InitiatingFlow``, the ``shareParentSessions`` boolean parameter of ``FlowLogic.subFlow`` has been
removed. This was an unfortunate parameter that unnecessarily exposed the inner workings of flow sessions. Now, if
your sub-flow can be started outside the context of the parent flow then annotate it with ``@InitiatingFlow``. If
it's meant to be used as a continuation of the existing parent flow, such as ``CollectSignaturesFlow``, then it
doesn't need any annotation.
* Also related to ``InitiatingFlow``, the ``shareParentSessions`` boolean parameter of ``FlowLogic.subFlow`` has been
removed. Its purpose was to allow subflows to be inlined with the parent flow - i.e. the subflow does not initiate
new sessions with parties the parent flow has already started. This allowed flows to be used as building blocks. To
achieve the same effect now simply requires the subflow to be *not* annotated wth ``InitiatingFlow`` (i.e. we've made
this the default behaviour). If the subflow is not meant to be inlined, and is supposed to initiate flows on the
other side, the annotation is required.
* The ``InitiatingFlow`` annotation also has an integer ``version`` property which assigns the initiating flow a version
number, defaulting to 1 if it's not specified. This enables versioning of flows with nodes only accepting communication
if the version number matches. At some point we will support the ability for a node to have multiple versions of the
same flow registered, enabling backwards compatibility of flows.
* ``ContractUpgradeFlow.Instigator`` has been renamed to just ``ContractUpgradeFlow``.
@ -50,10 +46,6 @@ UNRELEASED
* Names of parties are now stored as a ``X500Name`` rather than a ``String``, to correctly enforce basic structure of the
name. As a result all node legal names must now be structured as X.500 distinguished names.
* The Bouncy Castle library ``X509CertificateHolder`` class is now used in place of ``X509Certificate`` in order to
have a consistent class used internally. Conversions to/from ``X509Certificate`` are done as required, but should
be avoided where possible.
* There are major changes to transaction signing in flows:
* You should use the new ``CollectSignaturesFlow`` and corresponding ``SignTransactionFlow`` which handle most
@ -78,11 +70,16 @@ UNRELEASED
* The original ``KeyPair`` signing methods have been left on the ``TransactionBuilder`` and ``SignedTransaction``, but
should only be used as part of unit testing.
* The ``InitiatingFlow`` annotation also has an integer ``version`` property which assigns the initiating flow a version
number, defaulting to 1 if it's specified. The flow version is included in the flow session request and the counterparty
will only respond and start their own flow if the version number matches to the one they've registered with. At some
point we will support the ability for a node to have multiple versions of the same flow registered, enabling backwards
compatibility of CorDapp flows.
* ``Timestamp`` used for validation/notarization time-range has been renamed to ``TimeWindow``.
There are now 4 factory methods ``TimeWindow.fromOnly(fromTime: Instant)``,
``TimeWindow.untilOnly(untilTime: Instant)``, ``TimeWindow.between(fromTime: Instant, untilTime: Instant)`` and
``TimeWindow.withTolerance(time: Instant, tolerance: Duration)``.
Previous constructors ``TimeWindow(fromTime: Instant, untilTime: Instant)`` and
``TimeWindow(time: Instant, tolerance: Duration)`` have been removed.
* The Bouncy Castle library ``X509CertificateHolder`` class is now used in place of ``X509Certificate`` in order to
have a consistent class used internally. Conversions to/from ``X509Certificate`` are done as required, but should
be avoided where possible.
* The certificate hierarchy has been changed in order to allow corda node to sign keys with proper certificate chain.
* The corda node will now be issued a restricted client CA for identity/transaction key signing.

35
docs/source/corda-plugins.rst
View File

@ -45,40 +45,7 @@ extensions to be created, or registered at startup. In particular:
jars. These static serving directories will not be available if the
bundled web server is not started.
c. The ``servicePlugins`` property returns a list of classes which will
be instantiated once during the ``AbstractNode.start`` call. These
classes must provide a single argument constructor which will receive a
``PluginServiceHub`` reference. They must also extend the abstract class
``SingletonSerializeAsToken`` which ensures that if any reference to your
service is captured in a flow checkpoint (i.e. serialized by Kryo as
part of Quasar checkpoints, either on the stack or by reference within
your flows) it is stored as a simple token representing your service.
When checkpoints are restored, after a node restart for example,
the latest instance of the service will be substituted back in place of
the token stored in the checkpoint.
i. Firstly, they can call ``PluginServiceHub.registerServiceFlow`` and
register flows that will be initiated locally in response to remote flow
requests.
ii. Second, the service can hold a long lived reference to the
PluginServiceHub and to other private data, so the service can be used
to provide Oracle functionality. This Oracle functionality would
typically be exposed to other nodes by flows which are given a reference
to the service plugin when initiated (as defined by the
``registerServiceFlow`` call). The flow can then call into functions
on the plugin service singleton. Note, care should be taken to not allow
flows to hold references to fields which are not
also ``SingletonSerializeAsToken``, otherwise Quasar suspension in the
``StateMachineManager`` will fail with exceptions. An example oracle can
be seen in ``NodeInterestRates.kt`` in the irs-demo sample.
iii. The final use case for service plugins is that they can spawn threads, or register
to monitor vault updates. This allows them to provide long lived active
functions inside the node, for instance to initiate workflows when
certain conditions are met.
d. The ``customizeSerialization`` function allows classes to be whitelisted
c. The ``customizeSerialization`` function allows classes to be whitelisted
for object serialisation, over and above those tagged with the ``@CordaSerializable``
annotation. In general the annotation should be preferred. For
instance new state types will need to be explicitly registered. This will be called at

27
docs/source/creating-a-cordapp.rst
View File

@ -12,20 +12,33 @@ App plugins
To create an app plugin you must extend from `CordaPluginRegistry`_. The JavaDoc contains
specific details of the implementation, but you can extend the server in the following ways:
1. Service plugins: Register your services (see below).
1. Register your flows and services (see below).
2. Web APIs: You may register your own endpoints under /api/ of the bundled web server.
3. Static web endpoints: You may register your own static serving directories for serving web content from the web server.
4. Whitelisting your additional contract, state and other classes for object serialization. Any class that forms part
of a persisted state, that is used in messaging between flows or in RPC needs to be whitelisted.
Services
--------
Flows and services
------------------
Services are classes which are constructed after the node has started. It is provided a `PluginServiceHub`_ which
allows a richer API than the `ServiceHub`_ exposed to contracts. It enables adding flows, registering
message handlers and more. The service does not run in a separate thread, so the only entry point to the service is during
construction, where message handlers should be registered and threads started.
Flows are of two types: initiating and initiated. Initiating flows need to be annotated with ``@InitiatingFlow`` and can
be started in one of three ways:
1. By a user of your CorDapp via RPC in which the flow also needs to be annotated with ``@StartableByRPC``.
2. By another CorDapp executing it as a sub-flow in their own flow.
3. By a ``SchedulableState`` activity event, in which the flow also needs to be annotated with ``@SchedulableFlow``
``InitiatingFlow`` also has a ``version`` property to enable you to version your flows. A node will only accept communication
from an initiating party if the version numbers match up.
Initiated flows are typically private to your CorDapp and need to be annotated with ``@InitiatedBy`` which point to
initiating flow Class. The node scans your CorDapps for these annotations and automatically registers the initiating to
initiated mapping for you.
If your CorDapp also needs to have additional services running in the node, such as oracles, then annotate your service
class with ``@CordaService``. As with the flows, the node will automatically register it and make it available for use by
your flows. The service class has to implement ``SerializeAsToken`` to ensure they work correctly within flows. If possible
extend ``SingletonSerializeAsToken`` instead to avoid the boilerplate.
Starting nodes
--------------

10
docs/source/example-code/src/main/kotlin/net/corda/docs/FxTransactionBuildTutorial.kt
View File

@ -9,9 +9,9 @@ import net.corda.core.contracts.TransactionType
import net.corda.core.crypto.DigitalSignature
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.identity.Party
import net.corda.core.node.PluginServiceHub
import net.corda.core.node.ServiceHub
import net.corda.core.node.services.unconsumedStates
import net.corda.core.serialization.CordaSerializable
@ -21,13 +21,6 @@ import net.corda.flows.FinalityFlow
import net.corda.flows.ResolveTransactionsFlow
import java.util.*
object FxTransactionDemoTutorial {
// Would normally be called by custom service init in a CorDapp
fun registerFxProtocols(pluginHub: PluginServiceHub) {
pluginHub.registerServiceFlow(ForeignExchangeFlow::class.java, ::ForeignExchangeRemoteFlow)
}
}
@CordaSerializable
private data class FxRequest(val tradeId: String,
val amount: Amount<Issued<Currency>>,
@ -212,6 +205,7 @@ class ForeignExchangeFlow(val tradeId: String,
// DOCEND 3
}
@InitiatedBy(ForeignExchangeFlow::class)
class ForeignExchangeRemoteFlow(val source: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() {

22
docs/source/example-code/src/main/kotlin/net/corda/docs/WorkflowTransactionBuildTutorial.kt
View File

@ -6,10 +6,10 @@ import net.corda.core.crypto.DigitalSignature
import net.corda.core.crypto.SecureHash
import net.corda.core.crypto.containsAny
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.identity.AbstractParty
import net.corda.core.identity.Party
import net.corda.core.node.PluginServiceHub
import net.corda.core.node.ServiceHub
import net.corda.core.node.services.linearHeadsOfType
import net.corda.core.serialization.CordaSerializable
@ -19,22 +19,13 @@ import net.corda.flows.FinalityFlow
import java.security.PublicKey
import java.time.Duration
object WorkflowTransactionBuildTutorial {
// Would normally be called by custom service init in a CorDapp
fun registerWorkflowProtocols(pluginHub: PluginServiceHub) {
pluginHub.registerServiceFlow(SubmitCompletionFlow::class.java, ::RecordCompletionFlow)
}
}
// DOCSTART 1
// Helper method to locate the latest Vault version of a LinearState from a possibly out of date StateRef
inline fun <reified T : LinearState> ServiceHub.latest(ref: StateRef): StateAndRef<T> {
val linearHeads = vaultService.linearHeadsOfType<T>()
val original = toStateAndRef<T>(ref)
return linearHeads.get(original.state.data.linearId)!!
return linearHeads[original.state.data.linearId]!!
}
// DOCEND 1
// Minimal state model of a manual approval process
@ -87,7 +78,7 @@ data class TradeApprovalContract(override val legalContractReference: SecureHash
"Issue of new WorkflowContract must not include any inputs" using (tx.inputs.isEmpty())
"Issue of new WorkflowContract must be in a unique transaction" using (tx.outputs.size == 1)
}
val issued = tx.outputs.get(0) as TradeApprovalContract.State
val issued = tx.outputs[0] as TradeApprovalContract.State
requireThat {
"Issue requires the source Party as signer" using (command.signers.contains(issued.source.owningKey))
"Initial Issue state must be NEW" using (issued.state == WorkflowState.NEW)
@ -96,9 +87,9 @@ data class TradeApprovalContract(override val legalContractReference: SecureHash
is Commands.Completed -> {
val stateGroups = tx.groupStates(TradeApprovalContract.State::class.java) { it.linearId }
require(stateGroups.size == 1) { "Must be only a single proposal in transaction" }
for (group in stateGroups) {
val before = group.inputs.single()
val after = group.outputs.single()
for ((inputs, outputs) in stateGroups) {
val before = inputs.single()
val after = outputs.single()
requireThat {
"Only a non-final trade can be modified" using (before.state == WorkflowState.NEW)
"Output must be a final state" using (after.state in setOf(WorkflowState.APPROVED, WorkflowState.REJECTED))
@ -227,6 +218,7 @@ class SubmitCompletionFlow(val ref: StateRef, val verdict: WorkflowState) : Flow
* Then after checking to sign it and eventually store the fully notarised
* transaction to the ledger.
*/
@InitiatedBy(SubmitCompletionFlow::class)
class RecordCompletionFlow(val source: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call(): Unit {

7
docs/source/example-code/src/test/kotlin/net/corda/docs/FxTransactionBuildTutorialTest.kt
View File

@ -29,14 +29,11 @@ class FxTransactionBuildTutorialTest {
val notaryService = ServiceInfo(ValidatingNotaryService.type)
notaryNode = net.createNode(
legalName = DUMMY_NOTARY.name,
overrideServices = mapOf(Pair(notaryService, DUMMY_NOTARY_KEY)),
overrideServices = mapOf(notaryService to DUMMY_NOTARY_KEY),
advertisedServices = *arrayOf(ServiceInfo(NetworkMapService.type), notaryService))
nodeA = net.createPartyNode(notaryNode.info.address)
nodeB = net.createPartyNode(notaryNode.info.address)
FxTransactionDemoTutorial.registerFxProtocols(nodeA.services)
FxTransactionDemoTutorial.registerFxProtocols(nodeB.services)
WorkflowTransactionBuildTutorial.registerWorkflowProtocols(nodeA.services)
WorkflowTransactionBuildTutorial.registerWorkflowProtocols(nodeB.services)
nodeB.registerInitiatedFlow(ForeignExchangeRemoteFlow::class.java)
}
@After

8
docs/source/example-code/src/test/kotlin/net/corda/docs/WorkflowTransactionBuildTutorialTest.kt
View File

@ -4,7 +4,6 @@ import net.corda.core.contracts.LinearState
import net.corda.core.contracts.StateAndRef
import net.corda.core.contracts.StateRef
import net.corda.core.getOrThrow
import net.corda.core.node.ServiceEntry
import net.corda.core.node.ServiceHub
import net.corda.core.node.services.ServiceInfo
import net.corda.core.node.services.linearHeadsOfType
@ -30,7 +29,7 @@ class WorkflowTransactionBuildTutorialTest {
private inline fun <reified T : LinearState> ServiceHub.latest(ref: StateRef): StateAndRef<T> {
val linearHeads = vaultService.linearHeadsOfType<T>()
val original = storageService.validatedTransactions.getTransaction(ref.txhash)!!.tx.outRef<T>(ref.index)
return linearHeads.get(original.state.data.linearId)!!
return linearHeads[original.state.data.linearId]!!
}
@Before
@ -43,10 +42,7 @@ class WorkflowTransactionBuildTutorialTest {
advertisedServices = *arrayOf(ServiceInfo(NetworkMapService.type), notaryService))
nodeA = net.createPartyNode(notaryNode.info.address)
nodeB = net.createPartyNode(notaryNode.info.address)
FxTransactionDemoTutorial.registerFxProtocols(nodeA.services)
FxTransactionDemoTutorial.registerFxProtocols(nodeB.services)
WorkflowTransactionBuildTutorial.registerWorkflowProtocols(nodeA.services)
WorkflowTransactionBuildTutorial.registerWorkflowProtocols(nodeB.services)
nodeA.registerInitiatedFlow(RecordCompletionFlow::class.java)
}
@After

54
docs/source/flow-state-machines.rst
View File

@ -419,48 +419,56 @@ sequence of message transfers. Flows end pre-maturely due to exceptions, and as
Taking a step back, we mentioned that the other side has to accept the session request for there to be a communication
channel. A node accepts a session request if it has registered the flow type (the fully-qualified class name) that is
making the request - each session initiation includes the initiating flow type. The registration is done by a CorDapp
which has made available the particular flow communication, using ``PluginServiceHub.registerServiceFlow``. This method
specifies a flow factory for generating the counter-flow to any given initiating flow. If this registration doesn't exist
then no further communication takes place and the initiating flow ends with an exception. The initiating flow has to be
annotated with ``InitiatingFlow``.
making the request - each session initiation includes the initiating flow type. This registration is done automatically
by the node at startup by searching for flows which are annotated with ``@InitiatedBy``. This annotation points to the
flow that is doing the initiating, and this flow must be annotated with ``@InitiatingFlow``. The ``InitiatedBy`` flow
must have a constructor which takes in a single parameter of type ``Party`` - this is the initiating party.
Going back to our buyer and seller flows, we need a way to initiate communication between the two. This is typically done
with one side started manually using the ``startFlowDynamic`` RPC and this initiates the counter-flow on the other side.
In this case it doesn't matter which flow is the initiator and which is the initiated, which is why neither ``Buyer`` nor
``Seller`` are annotated with ``InitiatingFlow``. For example, if we choose the seller side as the initiator then we need
to create a simple seller starter flow that has the annotation we need:
with one side started manually using the ``startFlowDynamic`` RPC and this initiates the flow on the other side. In our
case it doesn't matter which flow is the initiator and which is the initiated, which is why neither ``Buyer`` nor ``Seller``
are annotated with ``InitiatedBy`` or ``InitiatingFlow``. If we, for example, choose the seller side as the initiator then
we need to create a simple seller starter flow that has the annotation we need:
.. container:: codeset
.. sourcecode:: kotlin
@InitiatingFlow
class SellerStarter(val otherParty: Party, val assetToSell: StateAndRef<OwnableState>, val price: Amount<Currency>) : FlowLogic<SignedTransaction>() {
class SellerInitiator(val buyer: Party,
val notary: NodeInfo,
val assetToSell: StateAndRef<OwnableState>,
val price: Amount<Currency>) : FlowLogic<SignedTransaction>() {
@Suspendable
override fun call(): SignedTransaction {
val notary: NodeInfo = serviceHub.networkMapCache.notaryNodes[0]
val cpOwnerKey: PublicKey = serviceHub.legalIdentityKey
return subFlow(TwoPartyTradeFlow.Seller(otherParty, notary, assetToSell, price, cpOwnerKey))
send(buyer, Pair(notary.notaryIdentity, price))
return subFlow(Seller(
buyer,
notary,
assetToSell,
price,
serviceHub.legalIdentityKey))
}
}
The buyer side would then need to register their flow, perhaps with something like:
The buyer side would look something like this. Notice the constructor takes in a single ``Party`` object which represents
the seller.
.. container:: codeset
.. sourcecode:: kotlin
val services: PluginServiceHub = TODO()
services.registerServiceFlow(SellerStarter::class.java) { otherParty ->
val notary = services.networkMapCache.notaryNodes[0]
val acceptablePrice = TODO()
val typeToBuy = TODO()
Buyer(otherParty, notary, acceptablePrice, typeToBuy)
@InitiatedBy(SellerInitiator::class)
class BuyerAcceptor(val seller: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() {
val (notary, price) = receive<Pair<Party, Amount<Currency>>>(seller).unwrap {
require(serviceHub.networkMapCache.isNotary(it.first)) { "${it.first} is not a notary" }
it
}
subFlow(Buyer(seller, notary, price, CommercialPaper.State::class.java))
}
}
This is telling the buyer node to fire up an instance of ``Buyer`` (the code in the lambda) when the initiating flow
is a seller (``SellerStarter::class.java``).
.. _progress-tracking:

42
docs/source/oracles.rst
View File

@ -195,41 +195,37 @@ Here we can see that there are several steps:
exactly our data source. The final step, assuming we have got this far, is to generate a signature for the
transaction and return it.
Binding to the network via a CorDapp plugin
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Binding to the network
~~~~~~~~~~~~~~~~~~~~~~
.. note:: Before reading any further, we advise that you understand the concept of flows and how to write them and use
them. See :doc:`flow-state-machines`. Likewise some understanding of Cordapps, plugins and services will be helpful.
See :doc:`creating-a-cordapp`.
The first step is to create a service to host the oracle on the network. Let's see how that's implemented:
The first step is to create the oracle as a service by annotating its class with ``@CordaService``. Let's see how that's
done:
.. literalinclude:: ../../samples/irs-demo/src/main/kotlin/net/corda/irs/api/NodeInterestRates.kt
:language: kotlin
:start-after: DOCSTART 3
:end-before: DOCEND 3
The Corda node scans for any class with this annotation and initialises them. The only requirement is that the class provide
a constructor with a single parameter of type ``PluginServiceHub```. In our example the oracle class has two constructors.
The second is used for testing.
.. literalinclude:: ../../samples/irs-demo/src/main/kotlin/net/corda/irs/api/NodeInterestRates.kt
:language: kotlin
:start-after: DOCSTART 2
:end-before: DOCEND 2
This may look complicated, but really it's made up of some relatively simple elements (in the order they appear in the code):
These two flows leverage the oracle to provide the querying and signing operations. They get reference to the oracle,
which will have already been initialised by the node, using ``ServiceHub.cordappService``. Both flows are annotated with
``@InitiatedBy``. This tells the node which initiating flow (which are discussed in the next section) they are meant to
be executed with.
1. Accept a ``PluginServiceHub`` in the constructor. This is your interface to the Corda node.
2. Ensure you extend the abstract class ``SingletonSerializeAsToken`` (see :doc:`corda-plugins`).
3. Create an instance of your core oracle class that has the ``query`` and ``sign`` methods as discussed above.
4. Register your client sub-flows (in this case both in ``RatesFixFlow``. See the next section) for querying and
signing as initiating your service flows that actually do the querying and signing using your core oracle class instance.
5. Implement your service flows that call your core oracle class instance.
The final step is to register your service with the node via the plugin mechanism. Do this by
implementing a plugin. Don't forget the resources file to register it with the ``ServiceLoader`` framework
(see :doc:`corda-plugins`).
.. sourcecode:: kotlin
class Plugin : CordaPluginRegistry() {
override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
}
Providing client sub-flows for querying and signing
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Providing sub-flows for querying and signing
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
We mentioned the client sub-flow briefly above. They are the mechanism that clients, in the form of other flows, will
interact with your oracle. Typically there will be one for querying and one for signing. Let's take a look at

8
docs/source/release-notes.rst
View File

@ -6,13 +6,15 @@ Here are release notes for each snapshot release from M9 onwards.
Unreleased
----------
We've added the ability for flows to be versioned by their CorDapp developers. This enables a node to support a particular
version of a flow and allows it to reject flow communication with a node which isn't using the same fact. In a future
release we allow a node to have multiple versions of the same flow running to enable backwards compatibility.
Writing CorDapps has been made simpler by removing boiler-plate code that was previously required when registering flows.
Instead we now make use of classpath scanning to automatically wire-up flows.
There are major changes to the ``Party`` class as part of confidential identities, and how parties and keys are stored
in transaction state objects. See :doc:`changelog` for full details.
We've added the ability for flows to be versioned by their CorDapp developers. This enables a node to support a particular
version of a flow and allows it to reject flow communication with a node which isn't using the same fact. In a future
release we allow a node to have multiple versions of the same flow running to enable backwards compatibility.
Milestone 11
------------

13
finance/src/main/kotlin/net/corda/flows/IssuerFlow.kt
View File

@ -2,12 +2,8 @@ package net.corda.flows
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.contracts.*
import net.corda.core.flows.FlowException
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.*
import net.corda.core.identity.Party
import net.corda.core.flows.StartableByRPC
import net.corda.core.node.PluginServiceHub
import net.corda.core.serialization.CordaSerializable
import net.corda.core.serialization.OpaqueBytes
import net.corda.core.transactions.SignedTransaction
@ -46,6 +42,7 @@ object IssuerFlow {
* Issuer refers to a Node acting as a Bank Issuer of [FungibleAsset], and processes requests from a [IssuanceRequester] client.
* Returns the generated transaction representing the transfer of the [Issued] [FungibleAsset] to the issue requester.
*/
@InitiatedBy(IssuanceRequester::class)
class Issuer(val otherParty: Party) : FlowLogic<SignedTransaction>() {
companion object {
object AWAITING_REQUEST : ProgressTracker.Step("Awaiting issuance request")
@ -97,11 +94,5 @@ object IssuerFlow {
// NOTE: CashFlow PayCash calls FinalityFlow which performs a Broadcast (which stores a local copy of the txn to the ledger)
return moveTx
}
class Service(services: PluginServiceHub) {
init {
services.registerServiceFlow(IssuanceRequester::class.java, ::Issuer)
}
}
}
}

17
finance/src/test/kotlin/net/corda/flows/IssuerFlowTest.kt
View File

@ -11,16 +11,17 @@ import net.corda.core.getOrThrow
import net.corda.core.identity.Party
import net.corda.core.map
import net.corda.core.serialization.OpaqueBytes
import net.corda.core.toFuture
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.DUMMY_NOTARY
import net.corda.flows.IssuerFlow.IssuanceRequester
import net.corda.testing.BOC
import net.corda.testing.MEGA_CORP
import net.corda.testing.initiateSingleShotFlow
import net.corda.testing.ledger
import net.corda.testing.node.MockNetwork
import net.corda.testing.node.MockNetwork.MockNode
import org.junit.Test
import rx.Observable
import java.util.*
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
@ -73,7 +74,6 @@ class IssuerFlowTest {
@Test
fun `test concurrent issuer flow`() {
net = MockNetwork(false, true)
ledger {
notaryNode = net.createNotaryNode(null, DUMMY_NOTARY.name)
@ -96,18 +96,19 @@ class IssuerFlowTest {
}
}
private fun runIssuerAndIssueRequester(issuerNode: MockNode, issueToNode: MockNode,
private fun runIssuerAndIssueRequester(issuerNode: MockNode,
issueToNode: MockNode,
amount: Amount<Currency>,
party: Party, ref: OpaqueBytes): RunResult {
party: Party,
ref: OpaqueBytes): RunResult {
val issueToPartyAndRef = party.ref(ref)
val issuerFuture = issuerNode.initiateSingleShotFlow(IssuerFlow.IssuanceRequester::class) { _ ->
IssuerFlow.Issuer(party)
}.map { it.stateMachine }
val issuerFlows: Observable<IssuerFlow.Issuer> = issuerNode.registerInitiatedFlow(IssuerFlow.Issuer::class.java)
val firstIssuerFiber = issuerFlows.toFuture().map { it.stateMachine }
val issueRequest = IssuanceRequester(amount, party, issueToPartyAndRef.reference, issuerNode.info.legalIdentity)
val issueRequestResultFuture = issueToNode.services.startFlow(issueRequest).resultFuture
return IssuerFlowTest.RunResult(issuerFuture, issueRequestResultFuture)
return IssuerFlowTest.RunResult(firstIssuerFiber, issueRequestResultFuture)
}
private data class RunResult(

10
node/build.gradle
View File

@ -46,6 +46,14 @@ processResources {
from file("$rootDir/config/dev/log4j2.xml")
}
processIntegrationTestResources {
// Build one of the demos so that we can test CorDapp scanning in CordappScanningTest. It doesn't matter which demo
// we use, just make sure the test is updated accordingly.
from(project(':samples:trader-demo').tasks.jar) {
rename 'trader-demo-(.*)', 'trader-demo.jar'
}
}
// To find potential version conflicts, run "gradle htmlDependencyReport" and then look in
// build/reports/project/dependencies/index.html for green highlighted parts of the tree.
@ -153,7 +161,7 @@ dependencies {
compile "io.requery:requery-kotlin:$requery_version"
// FastClasspathScanner: classpath scanning
compile 'io.github.lukehutch:fast-classpath-scanner:2.0.20'
compile 'io.github.lukehutch:fast-classpath-scanner:2.0.21'
// Integration test helpers
integrationTestCompile "junit:junit:$junit_version"

82
node/src/integration-test/kotlin/net/corda/node/CordappScanningTest.kt Normal file
View File

@ -0,0 +1,82 @@
package net.corda.node
import co.paralleluniverse.fibers.Suspendable
import com.google.common.util.concurrent.Futures
import net.corda.core.copyToDirectory
import net.corda.core.createDirectories
import net.corda.core.div
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StartableByRPC
import net.corda.core.getOrThrow
import net.corda.core.identity.Party
import net.corda.core.messaging.startFlow
import net.corda.core.utilities.ALICE
import net.corda.core.utilities.BOB
import net.corda.core.utilities.unwrap
import net.corda.node.driver.driver
import net.corda.node.services.startFlowPermission
import net.corda.nodeapi.User
import org.assertj.core.api.Assertions.assertThat
import org.junit.Test
import java.nio.file.Paths
class CordappScanningTest {
@Test
fun `CorDapp jar in plugins directory is scanned`() {
// If the CorDapp jar does't exist then run the integrationTestClasses gradle task
val cordappJar = Paths.get(javaClass.getResource("/trader-demo.jar").toURI())
driver {
val pluginsDir = (baseDirectory(ALICE.name) / "plugins").createDirectories()
cordappJar.copyToDirectory(pluginsDir)
val user = User("u", "p", emptySet())
val alice = startNode(ALICE.name, rpcUsers = listOf(user)).getOrThrow()
val rpc = alice.rpcClientToNode().start(user.username, user.password)
// If the CorDapp wasn't scanned then SellerFlow won't have been picked up as an RPC flow
assertThat(rpc.proxy.registeredFlows()).contains("net.corda.traderdemo.flow.SellerFlow")
}
}
@Test
fun `empty plugins directory`() {
driver {
val baseDirectory = baseDirectory(ALICE.name)
(baseDirectory / "plugins").createDirectories()
startNode(ALICE.name).getOrThrow()
}
}
@Test
fun `sub-classed initiated flow pointing to the same initiating flow as its super-class`() {
val user = User("u", "p", setOf(startFlowPermission<ReceiveFlow>()))
driver(systemProperties = mapOf("net.corda.node.cordapp.scan.package" to "net.corda.node")) {
val (alice, bob) = Futures.allAsList(
startNode(ALICE.name, rpcUsers = listOf(user)),
startNode(BOB.name)).getOrThrow()
val initiatedFlowClass = alice.rpcClientToNode()
.start(user.username, user.password)
.proxy
.startFlow(::ReceiveFlow, bob.nodeInfo.legalIdentity)
.returnValue
assertThat(initiatedFlowClass.getOrThrow()).isEqualTo(SendSubClassFlow::class.java.name)
}
}
@StartableByRPC
@InitiatingFlow
class ReceiveFlow(val otherParty: Party) : FlowLogic<String>() {
@Suspendable
override fun call(): String = receive<String>(otherParty).unwrap { it }
}
@InitiatedBy(ReceiveFlow::class)
open class SendClassFlow(val otherParty: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() = send(otherParty, javaClass.name)
}
@InitiatedBy(ReceiveFlow::class)
class SendSubClassFlow(otherParty: Party) : SendClassFlow(otherParty)
}

4
node/src/integration-test/kotlin/net/corda/services/messaging/MQSecurityTest.kt
View File

@ -6,6 +6,7 @@ import net.corda.client.rpc.CordaRPCClient
import net.corda.core.crypto.generateKeyPair
import net.corda.core.crypto.toBase58String
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.getOrThrow
import net.corda.core.identity.Party
@ -216,7 +217,7 @@ abstract class MQSecurityTest : NodeBasedTest() {
private fun startBobAndCommunicateWithAlice(): Party {
val bob = startNode(BOB.name).getOrThrow()
bob.services.registerServiceFlow(SendFlow::class.java, ::ReceiveFlow)
bob.registerInitiatedFlow(ReceiveFlow::class.java)
val bobParty = bob.info.legalIdentity
// Perform a protocol exchange to force the peer queue to be created
alice.services.startFlow(SendFlow(bobParty, 0)).resultFuture.getOrThrow()
@ -229,6 +230,7 @@ abstract class MQSecurityTest : NodeBasedTest() {
override fun call() = send(otherParty, payload)
}
@InitiatedBy(SendFlow::class)
private class ReceiveFlow(val otherParty: Party) : FlowLogic<Any>() {
@Suspendable
override fun call() = receive<Any>(otherParty).unwrap { it }

15
node/src/main/kotlin/net/corda/node/driver/Driver.kt
View File

@ -7,6 +7,8 @@ import com.google.common.util.concurrent.*
import com.typesafe.config.Config
import com.typesafe.config.ConfigRenderOptions
import net.corda.client.rpc.CordaRPCClient
import net.corda.cordform.CordformContext
import net.corda.cordform.CordformNode
import net.corda.core.*
import net.corda.core.crypto.X509Utilities
import net.corda.core.crypto.appendToCommonName
@ -19,10 +21,6 @@ import net.corda.core.node.services.ServiceType
import net.corda.core.utilities.*
import net.corda.node.LOGS_DIRECTORY_NAME
import net.corda.node.services.config.*
import net.corda.node.services.config.ConfigHelper
import net.corda.node.services.config.FullNodeConfiguration
import net.corda.node.services.config.VerifierType
import net.corda.node.services.config.configOf
import net.corda.node.services.network.NetworkMapService
import net.corda.node.services.transactions.RaftValidatingNotaryService
import net.corda.node.utilities.ServiceIdentityGenerator
@ -30,8 +28,6 @@ import net.corda.nodeapi.ArtemisMessagingComponent
import net.corda.nodeapi.User
import net.corda.nodeapi.config.SSLConfiguration
import net.corda.nodeapi.config.parseAs
import net.corda.cordform.CordformNode
import net.corda.cordform.CordformContext
import net.corda.core.internal.ShutdownHook
import net.corda.core.internal.addShutdownHook
import okhttp3.OkHttpClient
@ -39,6 +35,7 @@ import okhttp3.Request
import org.bouncycastle.asn1.x500.X500Name
import org.slf4j.Logger
import java.io.File
import java.io.File.pathSeparator
import java.net.*
import java.nio.file.Path
import java.nio.file.Paths
@ -614,7 +611,7 @@ class DriverDSL(
}
}
override fun baseDirectory(nodeName: X500Name) = driverDirectory / nodeName.commonName.replace(WHITESPACE, "")
override fun baseDirectory(nodeName: X500Name): Path = driverDirectory / nodeName.commonName.replace(WHITESPACE, "")
override fun startDedicatedNetworkMapService(): ListenableFuture<Unit> {
val debugPort = if (isDebug) debugPortAllocation.nextPort() else null
@ -679,6 +676,8 @@ class DriverDSL(
"-javaagent:$quasarJarPath"
val loggingLevel = if (debugPort == null) "INFO" else "DEBUG"
val pluginsDirectory = nodeConf.baseDirectory / "plugins"
ProcessUtilities.startJavaProcess(
className = "net.corda.node.Corda", // cannot directly get class for this, so just use string
arguments = listOf(
@ -686,6 +685,8 @@ class DriverDSL(
"--logging-level=$loggingLevel",
"--no-local-shell"
),
// Like the capsule, include the node's plugin directory
classpath = "${ProcessUtilities.defaultClassPath}$pathSeparator$pluginsDirectory/*",
jdwpPort = debugPort,
extraJvmArguments = extraJvmArguments,
errorLogPath = nodeConf.baseDirectory / LOGS_DIRECTORY_NAME / "error.log",

284
node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
View File

@ -2,16 +2,15 @@ package net.corda.node.internal
import com.codahale.metrics.MetricRegistry
import com.google.common.annotations.VisibleForTesting
import com.google.common.collect.MutableClassToInstanceMap
import com.google.common.util.concurrent.ListenableFuture
import com.google.common.util.concurrent.MoreExecutors
import com.google.common.util.concurrent.SettableFuture
import io.github.lukehutch.fastclasspathscanner.FastClasspathScanner
import io.github.lukehutch.fastclasspathscanner.scanner.ScanResult
import net.corda.core.*
import net.corda.core.crypto.*
import net.corda.core.flows.FlowInitiator
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StartableByRPC
import net.corda.core.flows.*
import net.corda.core.identity.Party
import net.corda.core.messaging.CordaRPCOps
import net.corda.core.messaging.RPCOps
@ -19,6 +18,7 @@ import net.corda.core.messaging.SingleMessageRecipient
import net.corda.core.node.*
import net.corda.core.node.services.*
import net.corda.core.node.services.NetworkMapCache.MapChange
import net.corda.core.serialization.SerializeAsToken
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.serialization.deserialize
import net.corda.core.transactions.SignedTransaction
@ -45,7 +45,7 @@ import net.corda.node.services.schema.HibernateObserver
import net.corda.node.services.schema.NodeSchemaService
import net.corda.node.services.statemachine.FlowStateMachineImpl
import net.corda.node.services.statemachine.StateMachineManager
import net.corda.node.services.statemachine.flowVersion
import net.corda.node.services.statemachine.flowVersionAndInitiatingClass
import net.corda.node.services.transactions.*
import net.corda.node.services.vault.CashBalanceAsMetricsObserver
import net.corda.node.services.vault.NodeVaultService
@ -60,9 +60,11 @@ import org.bouncycastle.cert.X509CertificateHolder
import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter
import org.jetbrains.exposed.sql.Database
import org.slf4j.Logger
import rx.Observable
import java.io.IOException
import java.lang.reflect.Modifier.*
import java.net.URL
import java.net.JarURLConnection
import java.net.URI
import java.nio.file.FileAlreadyExistsException
import java.nio.file.Path
import java.nio.file.Paths
@ -73,6 +75,7 @@ import java.util.*
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ExecutorService
import java.util.concurrent.TimeUnit.SECONDS
import java.util.stream.Collectors.toList
import kotlin.collections.ArrayList
import kotlin.reflect.KClass
import net.corda.core.crypto.generateKeyPair as cryptoGenerateKeyPair
@ -106,7 +109,8 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
// low-performance prototyping period.
protected abstract val serverThread: AffinityExecutor
protected val serviceFlowFactories = ConcurrentHashMap<Class<*>, ServiceFlowInfo>()
private val cordappServices = MutableClassToInstanceMap.create<SerializeAsToken>()
private val flowFactories = ConcurrentHashMap<Class<out FlowLogic<*>>, InitiatedFlowFactory<*>>()
protected val partyKeys = mutableSetOf<KeyPair>()
val services = object : ServiceHubInternal() {
@ -122,6 +126,12 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
override val schemaService: SchemaService get() = schemas
override val transactionVerifierService: TransactionVerifierService get() = txVerifierService
override val auditService: AuditService get() = auditService
override fun <T : SerializeAsToken> cordaService(type: Class<T>): T {
require(type.isAnnotationPresent(CordaService::class.java)) { "${type.name} is not a Corda service" }
return cordappServices.getInstance(type) ?: throw IllegalArgumentException("Corda service ${type.name} does not exist")
}
override val rpcFlows: List<Class<out FlowLogic<*>>> get() = this@AbstractNode.rpcFlows
// Internal only
@ -131,17 +141,8 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
return serverThread.fetchFrom { smm.add(logic, flowInitiator) }
}
override fun registerServiceFlow(initiatingFlowClass: Class<out FlowLogic<*>>, serviceFlowFactory: (Party) -> FlowLogic<*>) {
require(initiatingFlowClass !in serviceFlowFactories) {
"${initiatingFlowClass.name} has already been used to register a service flow"
}
val info = ServiceFlowInfo.CorDapp(initiatingFlowClass.flowVersion, serviceFlowFactory)
log.info("Registering service flow for ${initiatingFlowClass.name}: $info")
serviceFlowFactories[initiatingFlowClass] = info
}
override fun getServiceFlowFactory(clientFlowClass: Class<out FlowLogic<*>>): ServiceFlowInfo? {
return serviceFlowFactories[clientFlowClass]
override fun getFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>): InitiatedFlowFactory<*>? {
return flowFactories[initiatingFlowClass]
}
override fun recordTransactions(txs: Iterable<SignedTransaction>) {
@ -167,15 +168,11 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
lateinit var scheduler: NodeSchedulerService
lateinit var schemas: SchemaService
lateinit var auditService: AuditService
val customServices: ArrayList<Any> = ArrayList()
protected val runOnStop: ArrayList<Runnable> = ArrayList()
lateinit var database: Database
protected var dbCloser: Runnable? = null
private lateinit var rpcFlows: List<Class<out FlowLogic<*>>>
/** Locates and returns a service of the given type if loaded, or throws an exception if not found. */
inline fun <reified T : Any> findService() = customServices.filterIsInstance<T>().single()
var isPreviousCheckpointsPresent = false
private set
@ -217,7 +214,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
val tokenizableServices = makeServices()
smm = StateMachineManager(services,
listOf(tokenizableServices),
checkpointStorage,
serverThread,
database,
@ -240,22 +236,24 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
startMessagingService(rpcOps)
installCoreFlows()
fun Class<out FlowLogic<*>>.isUserInvokable(): Boolean {
return isPublic(modifiers) && !isLocalClass && !isAnonymousClass && (!isMemberClass || isStatic(modifiers))
val scanResult = scanCorDapps()
if (scanResult != null) {
val cordappServices = installCordaServices(scanResult)
tokenizableServices.addAll(cordappServices)
registerInitiatedFlows(scanResult)
rpcFlows = findRPCFlows(scanResult)
} else {
rpcFlows = emptyList()
}
val flows = scanForFlows()
rpcFlows = flows.filter { it.isUserInvokable() && it.isAnnotationPresent(StartableByRPC::class.java) } +
// Add any core flows here
listOf(ContractUpgradeFlow::class.java,
// TODO Remove all Cash flows from default list once they are split into separate CorDapp.
CashIssueFlow::class.java,
CashExitFlow::class.java,
CashPaymentFlow::class.java)
// TODO Remove this once the cash stuff is in its own CorDapp
registerInitiatedFlow(IssuerFlow.Issuer::class.java)
initUploaders()
runOnStop += Runnable { net.stop() }
_networkMapRegistrationFuture.setFuture(registerWithNetworkMapIfConfigured())
smm.start()
smm.start(tokenizableServices)
// Shut down the SMM so no Fibers are scheduled.
runOnStop += Runnable { smm.stop(acceptableLiveFiberCountOnStop()) }
scheduler.start()
@ -264,18 +262,142 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
return this
}
private fun installCordaServices(scanResult: ScanResult): List<SerializeAsToken> {
return scanResult.getClassesWithAnnotation(SerializeAsToken::class, CordaService::class).mapNotNull {
try {
installCordaService(it)
} catch (e: NoSuchMethodException) {
log.error("${it.name}, as a Corda service, must have a constructor with a single parameter " +
"of type ${PluginServiceHub::class.java.name}")
null
} catch (e: Exception) {
log.error("Unable to install Corda service ${it.name}", e)
null
}
}
}
/**
* Use this method to install your Corda services in your tests. This is automatically done by the node when it
* starts up for all classes it finds which are annotated with [CordaService].
*/
fun <T : SerializeAsToken> installCordaService(clazz: Class<T>): T {
clazz.requireAnnotation<CordaService>()
val ctor = clazz.getDeclaredConstructor(PluginServiceHub::class.java).apply { isAccessible = true }
val service = ctor.newInstance(services)
cordappServices.putInstance(clazz, service)
log.info("Installed ${clazz.name} Corda service")
return service
}
private inline fun <reified A : Annotation> Class<*>.requireAnnotation(): A {
return requireNotNull(getDeclaredAnnotation(A::class.java)) { "$name needs to be annotated with ${A::class.java.name}" }
}
private fun registerInitiatedFlows(scanResult: ScanResult) {
scanResult
.getClassesWithAnnotation(FlowLogic::class, InitiatedBy::class)
// First group by the initiating flow class in case there are multiple mappings
.groupBy { it.requireAnnotation<InitiatedBy>().value.java }
.map { (initiatingFlow, initiatedFlows) ->
val sorted = initiatedFlows.sortedWith(FlowTypeHierarchyComparator(initiatingFlow))
if (sorted.size > 1) {
log.warn("${initiatingFlow.name} has been specified as the inititating flow by multiple flows " +
"in the same type hierarchy: ${sorted.joinToString { it.name }}. Choosing the most " +
"specific sub-type for registration: ${sorted[0].name}.")
}
sorted[0]
}
.forEach {
try {
registerInitiatedFlowInternal(it, track = false)
} catch (e: NoSuchMethodException) {
log.error("${it.name}, as an initiated flow, must have a constructor with a single parameter " +
"of type ${Party::class.java.name}")
} catch (e: Exception) {
log.error("Unable to register initiated flow ${it.name}", e)
}
}
}
private class FlowTypeHierarchyComparator(val initiatingFlow: Class<out FlowLogic<*>>) : Comparator<Class<out FlowLogic<*>>> {
override fun compare(o1: Class<out FlowLogic<*>>, o2: Class<out FlowLogic<*>>): Int {
return if (o1 == o2) {
0
} else if (o1.isAssignableFrom(o2)) {
1
} else if (o2.isAssignableFrom(o1)) {
-1
} else {
throw IllegalArgumentException("${initiatingFlow.name} has been specified as the initiating flow by " +
"both ${o1.name} and ${o2.name}")
}
}
}
/**
* Use this method to register your initiated flows in your tests. This is automatically done by the node when it
* starts up for all [FlowLogic] classes it finds which are annotated with [InitiatedBy].
* @return An [Observable] of the initiated flows started by counter-parties.
*/
fun <T : FlowLogic<*>> registerInitiatedFlow(initiatedFlowClass: Class<T>): Observable<T> {
return registerInitiatedFlowInternal(initiatedFlowClass, track = true)
}
private fun <F : FlowLogic<*>> registerInitiatedFlowInternal(initiatedFlow: Class<F>, track: Boolean): Observable<F> {
val ctor = initiatedFlow.getDeclaredConstructor(Party::class.java).apply { isAccessible = true }
val initiatingFlow = initiatedFlow.requireAnnotation<InitiatedBy>().value.java
val (version, classWithAnnotation) = initiatingFlow.flowVersionAndInitiatingClass
require(classWithAnnotation == initiatingFlow) {
"${InitiatingFlow::class.java.name} must be annotated on ${initiatingFlow.name} and not on a super-type"
}
val flowFactory = InitiatedFlowFactory.CorDapp(version, { ctor.newInstance(it) })
val observable = registerFlowFactory(initiatingFlow, flowFactory, initiatedFlow, track)
log.info("Registered ${initiatingFlow.name} to initiate ${initiatedFlow.name} (version $version)")
return observable
}
@VisibleForTesting
fun <F : FlowLogic<*>> registerFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>,
flowFactory: InitiatedFlowFactory<F>,
initiatedFlowClass: Class<F>,
track: Boolean): Observable<F> {
val observable = if (track) {
smm.changes.filter { it is StateMachineManager.Change.Add }.map { it.logic }.ofType(initiatedFlowClass)
} else {
Observable.empty()
}
flowFactories[initiatingFlowClass] = flowFactory
return observable
}
private fun findRPCFlows(scanResult: ScanResult): List<Class<out FlowLogic<*>>> {
fun Class<out FlowLogic<*>>.isUserInvokable(): Boolean {
return isPublic(modifiers) && !isLocalClass && !isAnonymousClass && (!isMemberClass || isStatic(modifiers))
}
return scanResult.getClassesWithAnnotation(FlowLogic::class, StartableByRPC::class).filter { it.isUserInvokable() } +
// Add any core flows here
listOf(
ContractUpgradeFlow::class.java,
// TODO Remove all Cash flows from default list once they are split into separate CorDapp.
CashIssueFlow::class.java,
CashExitFlow::class.java,
CashPaymentFlow::class.java)
}
/**
* Installs a flow that's core to the Corda platform. Unlike CorDapp flows which are versioned individually using
* [InitiatingFlow.version], core flows have the same version as the node's platform version. To cater for backwards
* compatibility [serviceFlowFactory] provides a second parameter which is the platform version of the initiating party.
* compatibility [flowFactory] provides a second parameter which is the platform version of the initiating party.
* @suppress
*/
@VisibleForTesting
fun installCoreFlow(clientFlowClass: KClass<out FlowLogic<*>>, serviceFlowFactory: (Party, Int) -> FlowLogic<*>) {
require(clientFlowClass.java.flowVersion == 1) {
fun installCoreFlow(clientFlowClass: KClass<out FlowLogic<*>>, flowFactory: (Party, Int) -> FlowLogic<*>) {
require(clientFlowClass.java.flowVersionAndInitiatingClass.first == 1) {
"${InitiatingFlow::class.java.name}.version not applicable for core flows; their version is the node's platform version"
}
serviceFlowFactories[clientFlowClass.java] = ServiceFlowInfo.Core(serviceFlowFactory)
flowFactories[clientFlowClass.java] = InitiatedFlowFactory.Core(flowFactory)
log.debug { "Installed core flow ${clientFlowClass.java.name}" }
}
@ -313,61 +435,62 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
val tokenizableServices = mutableListOf(storage, net, vault, keyManagement, identity, platformClock, scheduler)
makeAdvertisedServices(tokenizableServices)
customServices.clear()
customServices.addAll(makePluginServices(tokenizableServices))
initUploaders(storageServices)
return tokenizableServices
}
private fun scanForFlows(): List<Class<out FlowLogic<*>>> {
private fun scanCorDapps(): ScanResult? {
val scanPackage = System.getProperty("net.corda.node.cordapp.scan.package")
val paths = if (scanPackage != null) {
// This is purely for integration tests so that classes defined in the test can automatically be picked up
check(configuration.devMode) { "Package scanning can only occur in dev mode" }
val resource = scanPackage.replace('.', '/')
javaClass.classLoader.getResources(resource)
.asSequence()
.map {
val uri = if (it.protocol == "jar") {
(it.openConnection() as JarURLConnection).jarFileURL.toURI()
} else {
URI(it.toExternalForm().removeSuffix(resource))
}
Paths.get(uri)
}
.toList()
} else {
val pluginsDir = configuration.baseDirectory / "plugins"
log.info("Scanning plugins in $pluginsDir ...")
if (!pluginsDir.exists()) return emptyList()
val pluginJars = pluginsDir.list {
it.filter { it.isRegularFile() && it.toString().endsWith(".jar") }.toArray()
if (!pluginsDir.exists()) return null
pluginsDir.list {
it.filter { it.isRegularFile() && it.toString().endsWith(".jar") }.collect(toList())
}
}
if (pluginJars.isEmpty()) return emptyList()
log.info("Scanning CorDapps in $paths")
val scanResult = FastClasspathScanner().overrideClasspath(*pluginJars).scan() // This will only scan the plugin jars and nothing else
// This will only scan the plugin jars and nothing else
return if (paths.isNotEmpty()) FastClasspathScanner().overrideClasspath(paths).scan() else null
}
fun loadFlowClass(className: String): Class<out FlowLogic<*>>? {
private fun <T : Any> ScanResult.getClassesWithAnnotation(type: KClass<T>, annotation: KClass<out Annotation>): List<Class<out T>> {
fun loadClass(className: String): Class<out T>? {
return try {
// TODO Make sure this is loaded by the correct class loader
@Suppress("UNCHECKED_CAST")
Class.forName(className, false, javaClass.classLoader) as Class<out FlowLogic<*>>
Class.forName(className, false, javaClass.classLoader).asSubclass(type.java)
} catch (e: ClassCastException) {
log.warn("As $className is annotated with ${annotation.qualifiedName} it must be a sub-type of ${type.java.name}")
null
} catch (e: Exception) {
log.warn("Unable to load flow class $className", e)
log.warn("Unable to load class $className", e)
null
}
}
val flowClasses = scanResult.getNamesOfSubclassesOf(FlowLogic::class.java)
.mapNotNull { loadFlowClass(it) }
return getNamesOfClassesWithAnnotation(annotation.java)
.mapNotNull { loadClass(it) }
.filterNot { isAbstract(it.modifiers) }
fun URL.pluginName(): String {
return try {
Paths.get(toURI()).fileName.toString()
} catch (e: Exception) {
toString()
}
}
flowClasses.groupBy {
scanResult.classNameToClassInfo[it.name]!!.classpathElementURLs.first()
}.forEach { url, classes ->
log.info("Found flows in plugin ${url.pluginName()}: ${classes.joinToString { it.name }}")
}
return flowClasses
}
private fun initUploaders(storageServices: Pair<TxWritableStorageService, CheckpointStorage>) {
val uploaders: List<FileUploader> = listOf(storageServices.first.attachments as NodeAttachmentService) +
customServices.filterIsInstance(AcceptsFileUpload::class.java)
private fun initUploaders() {
val uploaders: List<FileUploader> = listOf(storage.attachments as NodeAttachmentService) +
cordappServices.values.filterIsInstance(AcceptsFileUpload::class.java)
(storage as StorageServiceImpl).initUploaders(uploaders)
}
@ -433,12 +556,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
}
}
private fun makePluginServices(tokenizableServices: MutableList<Any>): List<Any> {
val pluginServices = pluginRegistries.flatMap { it.servicePlugins }.map { it.apply(services) }
tokenizableServices.addAll(pluginServices)
return pluginServices
}
/**
* Run any tasks that are needed to ensure the node is in a correct state before running start().
*/
@ -658,11 +775,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
}
}
sealed class ServiceFlowInfo {
data class Core(val factory: (Party, Int) -> FlowLogic<*>) : ServiceFlowInfo()
data class CorDapp(val version: Int, val factory: (Party) -> FlowLogic<*>) : ServiceFlowInfo()
}
private class KeyStoreWrapper(private val storePath: Path, private val storePassword: String) {
private val keyStore = KeyStoreUtilities.loadKeyStore(storePath, storePassword)

27
node/src/main/kotlin/net/corda/node/internal/InitiatedFlowFactory.kt Normal file
View File

@ -0,0 +1,27 @@
package net.corda.node.internal
import net.corda.core.flows.FlowLogic
import net.corda.core.identity.Party
import net.corda.node.services.statemachine.SessionInit
interface InitiatedFlowFactory<out F : FlowLogic<*>> {
fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): F
data class Core<out F : FlowLogic<*>>(val factory: (Party, Int) -> F) : InitiatedFlowFactory<F> {
override fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): F {
return factory(otherParty, platformVersion)
}
}
data class CorDapp<out F : FlowLogic<*>>(val version: Int, val factory: (Party) -> F) : InitiatedFlowFactory<F> {
override fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): F {
// TODO Add support for multiple versions of the same flow when CorDapps are loaded in separate class loaders
if (sessionInit.flowVerison == version) return factory(otherParty)
throw SessionRejectException(
"Version not supported",
"Version mismatch - ${sessionInit.initiatingFlowClass} is only registered for version $version")
}
}
}
class SessionRejectException(val rejectMessage: String, val logMessage: String) : Exception()

7
node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt
View File

@ -13,7 +13,7 @@ import net.corda.core.node.services.TxWritableStorageService
import net.corda.core.serialization.CordaSerializable
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.loggerFor
import net.corda.node.internal.ServiceFlowInfo
import net.corda.node.internal.InitiatedFlowFactory
import net.corda.node.services.messaging.MessagingService
import net.corda.node.services.statemachine.FlowLogicRefFactoryImpl
import net.corda.node.services.statemachine.FlowStateMachineImpl
@ -47,7 +47,6 @@ interface NetworkMapCacheInternal : NetworkMapCache {
/** For testing where the network map cache is manipulated marks the service as immediately ready. */
@VisibleForTesting
fun runWithoutMapService()
}
@CordaSerializable
@ -93,7 +92,6 @@ abstract class ServiceHubInternal : PluginServiceHub {
* Starts an already constructed flow. Note that you must be on the server thread to call this method. [FlowInitiator]
* defaults to [FlowInitiator.RPC] with username "Only For Testing".
*/
// TODO Move it to test utils.
@VisibleForTesting
fun <T> startFlow(logic: FlowLogic<T>): FlowStateMachine<T> = startFlow(logic, FlowInitiator.RPC("Only For Testing"))
@ -103,7 +101,6 @@ abstract class ServiceHubInternal : PluginServiceHub {
*/
abstract fun <T> startFlow(logic: FlowLogic<T>, flowInitiator: FlowInitiator): FlowStateMachineImpl<T>
/**
* Will check [logicType] and [args] against a whitelist and if acceptable then construct and initiate the flow.
* Note that you must be on the server thread to call this method. [flowInitiator] points how flow was started,
@ -122,5 +119,5 @@ abstract class ServiceHubInternal : PluginServiceHub {
return startFlow(logic, flowInitiator)
}
abstract fun getServiceFlowFactory(clientFlowClass: Class<out FlowLogic<*>>): ServiceFlowInfo?
abstract fun getFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>): InitiatedFlowFactory<*>?
}

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

@ -8,9 +8,9 @@ import com.google.common.util.concurrent.ListenableFuture
import com.google.common.util.concurrent.SettableFuture
import net.corda.core.ErrorOr
import net.corda.core.abbreviate
import net.corda.core.identity.Party
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.*
import net.corda.core.identity.Party
import net.corda.core.random63BitValue
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.ProgressTracker
@ -27,7 +27,6 @@ import org.jetbrains.exposed.sql.Transaction
import org.jetbrains.exposed.sql.transactions.TransactionManager
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.lang.reflect.Modifier
import java.sql.Connection
import java.sql.SQLException
import java.util.*
@ -322,9 +321,8 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
logger.trace { "Initiating a new session with $otherParty" }
val session = FlowSession(sessionFlow, random63BitValue(), null, FlowSessionState.Initiating(otherParty), retryable)
openSessions[Pair(sessionFlow, otherParty)] = session
// We get the top-most concrete class object to cater for the case where the client flow is customised via a sub-class
val clientFlowClass = sessionFlow.topConcreteFlowClass
val sessionInit = SessionInit(session.ourSessionId, clientFlowClass, clientFlowClass.flowVersion, firstPayload)
val (version, initiatingFlowClass) = sessionFlow.javaClass.flowVersionAndInitiatingClass
val sessionInit = SessionInit(session.ourSessionId, initiatingFlowClass, version, firstPayload)
sendInternal(session, sessionInit)
if (waitForConfirmation) {
session.waitForConfirmation()
@ -332,15 +330,6 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
return session
}
@Suppress("UNCHECKED_CAST")
private val FlowLogic<*>.topConcreteFlowClass: Class<out FlowLogic<*>> get() {
var current: Class<out FlowLogic<*>> = javaClass
while (!Modifier.isAbstract(current.superclass.modifiers)) {
current = current.superclass as Class<out FlowLogic<*>>
}
return current
}
@Suspendable
private fun <M : ExistingSessionMessage> waitForMessage(receiveRequest: ReceiveRequest<M>): ReceivedSessionMessage<M> {
return receiveRequest.suspendAndExpectReceive().confirmReceiveType(receiveRequest)
@ -460,10 +449,19 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
}
}
val Class<out FlowLogic<*>>.flowVersion: Int get() {
val annotation = requireNotNull(getAnnotation(InitiatingFlow::class.java)) {
"$name as the initiating flow must be annotated with ${InitiatingFlow::class.java.name}"
}
@Suppress("UNCHECKED_CAST")
val Class<out FlowLogic<*>>.flowVersionAndInitiatingClass: Pair<Int, Class<out FlowLogic<*>>> get() {
var current: Class<*> = this
var found: Pair<Int, Class<out FlowLogic<*>>>? = null
while (true) {
val annotation = current.getDeclaredAnnotation(InitiatingFlow::class.java)
if (annotation != null) {
if (found != null) throw IllegalArgumentException("${InitiatingFlow::class.java.name} can only be annotated once")
require(annotation.version > 0) { "Flow versions have to be greater or equal to 1" }
return annotation.version
found = annotation.version to (current as Class<out FlowLogic<*>>)
}
current = current.superclass
?: return found
?: throw IllegalArgumentException("$name as an initiating flow must be annotated with ${InitiatingFlow::class.java.name}")
}
}

2
node/src/main/kotlin/net/corda/node/services/statemachine/SessionMessage.kt
View File

@ -15,7 +15,7 @@ import net.corda.core.utilities.UntrustworthyData
interface SessionMessage
data class SessionInit(val initiatorSessionId: Long,
val clientFlowClass: Class<out FlowLogic<*>>,
val initiatingFlowClass: Class<out FlowLogic<*>>,
val flowVerison: Int,
val firstPayload: Any?) : SessionMessage

39
node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineManager.kt
View File

@ -15,14 +15,14 @@ import com.google.common.collect.HashMultimap
import com.google.common.util.concurrent.ListenableFuture
import io.requery.util.CloseableIterator
import net.corda.core.*
import net.corda.core.identity.Party
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.*
import net.corda.core.identity.Party
import net.corda.core.serialization.*
import net.corda.core.utilities.debug
import net.corda.core.utilities.loggerFor
import net.corda.core.utilities.trace
import net.corda.node.internal.ServiceFlowInfo
import net.corda.node.internal.SessionRejectException
import net.corda.node.services.api.Checkpoint
import net.corda.node.services.api.CheckpointStorage
import net.corda.node.services.api.ServiceHubInternal
@ -61,7 +61,6 @@ import javax.annotation.concurrent.ThreadSafe
*/
@ThreadSafe
class StateMachineManager(val serviceHub: ServiceHubInternal,
tokenizableServices: List<Any>,
val checkpointStorage: CheckpointStorage,
val executor: AffinityExecutor,
val database: Database,
@ -147,7 +146,7 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
private val recentlyClosedSessions = ConcurrentHashMap<Long, Party>()
// Context for tokenized services in checkpoints
private val serializationContext = SerializeAsTokenContext(tokenizableServices, quasarKryoPool, serviceHub)
private lateinit var serializationContext: SerializeAsTokenContext
/** Returns a list of all state machines executing the given flow logic at the top level (subflows do not count) */
fun <P : FlowLogic<T>, T> findStateMachines(flowClass: Class<P>): List<Pair<P, ListenableFuture<T>>> {
@ -171,7 +170,8 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
*/
val changes: Observable<Change> = mutex.content.changesPublisher.wrapWithDatabaseTransaction()
fun start() {
fun start(tokenizableServices: List<Any>) {
serializationContext = SerializeAsTokenContext(tokenizableServices, quasarKryoPool, serviceHub)
restoreFibersFromCheckpoints()
listenToLedgerTransactions()
serviceHub.networkMapCache.mapServiceRegistered.then(executor) { resumeRestoredFibers() }
@ -345,28 +345,15 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
fun sendSessionReject(message: String) = sendSessionMessage(sender, SessionReject(otherPartySessionId, message))
val serviceFlowInfo = serviceHub.getServiceFlowFactory(sessionInit.clientFlowClass)
if (serviceFlowInfo == null) {
logger.warn("${sessionInit.clientFlowClass} has not been registered with a service flow: $sessionInit")
sendSessionReject("${sessionInit.clientFlowClass.name} has not been registered with a service flow")
val initiatedFlowFactory = serviceHub.getFlowFactory(sessionInit.initiatingFlowClass)
if (initiatedFlowFactory == null) {
logger.warn("${sessionInit.initiatingFlowClass} has not been registered: $sessionInit")
sendSessionReject("${sessionInit.initiatingFlowClass.name} has not been registered with a service flow")
return
}
val session = try {
val flow = when (serviceFlowInfo) {
is ServiceFlowInfo.CorDapp -> {
// TODO Add support for multiple versions of the same flow when CorDapps are loaded in separate class loaders
if (sessionInit.flowVerison != serviceFlowInfo.version) {
logger.warn("Version mismatch - ${sessionInit.clientFlowClass} is only registered for version " +
"${serviceFlowInfo.version}: $sessionInit")
sendSessionReject("Version not supported")
return
}
serviceFlowInfo.factory(sender)
}
is ServiceFlowInfo.Core -> serviceFlowInfo.factory(sender, receivedMessage.platformVersion)
}
val flow = initiatedFlowFactory.createFlow(receivedMessage.platformVersion, sender, sessionInit)
val fiber = createFiber(flow, FlowInitiator.Peer(sender))
val session = FlowSession(flow, random63BitValue(), sender, FlowSessionState.Initiated(sender, otherPartySessionId))
if (sessionInit.firstPayload != null) {
@ -376,6 +363,10 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
fiber.openSessions[Pair(flow, sender)] = session
updateCheckpoint(fiber)
session
} catch (e: SessionRejectException) {
logger.warn("${e.logMessage}: $sessionInit")
sendSessionReject(e.rejectMessage)
return
} catch (e: Exception) {
logger.warn("Couldn't start flow session from $sessionInit", e)
sendSessionReject("Unable to establish session")
@ -383,7 +374,7 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
}
sendSessionMessage(sender, SessionConfirm(otherPartySessionId, session.ourSessionId), session.fiber)
session.fiber.logger.debug { "Initiated by $sender using ${sessionInit.clientFlowClass.name}" }
session.fiber.logger.debug { "Initiated by $sender using ${sessionInit.initiatingFlowClass.name}" }
session.fiber.logger.trace { "Initiated from $sessionInit on $session" }
resumeFiber(session.fiber)
}

20
node/src/test/kotlin/net/corda/node/internal/NodeTest.kt
View File

@ -1,20 +0,0 @@
package net.corda.node.internal
import net.corda.core.createDirectories
import net.corda.core.crypto.commonName
import net.corda.core.div
import net.corda.core.getOrThrow
import net.corda.core.utilities.ALICE
import net.corda.core.utilities.WHITESPACE
import net.corda.testing.node.NodeBasedTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.Test
class NodeTest : NodeBasedTest() {
@Test
fun `empty plugins directory`() {
val baseDirectory = baseDirectory(ALICE.name)
(baseDirectory / "plugins").createDirectories()
startNode(ALICE.name).getOrThrow()
}
}

63
node/src/test/kotlin/net/corda/node/messaging/TwoPartyTradeFlowTests.kt
View File

@ -4,24 +4,18 @@ import co.paralleluniverse.fibers.Suspendable
import net.corda.contracts.CommercialPaper
import net.corda.contracts.asset.*
import net.corda.contracts.testing.fillWithSomeTestCash
import net.corda.core.*
import net.corda.core.contracts.*
import net.corda.core.crypto.DigitalSignature
import net.corda.core.crypto.SecureHash
import net.corda.core.crypto.sign
import net.corda.core.days
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.FlowStateMachine
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StateMachineRunId
import net.corda.core.getOrThrow
import net.corda.core.flows.*
import net.corda.core.identity.AbstractParty
import net.corda.core.identity.AnonymousParty
import net.corda.core.identity.Party
import net.corda.core.identity.AbstractParty
import net.corda.core.map
import net.corda.core.messaging.SingleMessageRecipient
import net.corda.core.node.NodeInfo
import net.corda.core.node.services.*
import net.corda.core.rootCause
import net.corda.core.serialization.serialize
import net.corda.core.transactions.SignedTransaction
import net.corda.core.transactions.TransactionBuilder
@ -86,9 +80,10 @@ class TwoPartyTradeFlowTests {
net = MockNetwork(false, true)
ledger {
val notaryNode = net.createNotaryNode(null, DUMMY_NOTARY.name)
val aliceNode = net.createPartyNode(notaryNode.info.address, ALICE.name)
val bobNode = net.createPartyNode(notaryNode.info.address, BOB.name)
val basketOfNodes = net.createSomeNodes(2)
val notaryNode = basketOfNodes.notaryNode
val aliceNode = basketOfNodes.partyNodes[0]
val bobNode = basketOfNodes.partyNodes[1]
aliceNode.disableDBCloseOnStop()
bobNode.disableDBCloseOnStop()
@ -137,8 +132,7 @@ class TwoPartyTradeFlowTests {
aliceNode.disableDBCloseOnStop()
bobNode.disableDBCloseOnStop()
val cashStates =
bobNode.database.transaction {
val cashStates = bobNode.database.transaction {
bobNode.services.fillWithSomeTestCash(2000.DOLLARS, notaryNode.info.notaryIdentity, 3, 3)
}
@ -239,7 +233,7 @@ class TwoPartyTradeFlowTests {
}, true, BOB.name)
// Find the future representing the result of this state machine again.
val bobFuture = bobNode.smm.findStateMachines(Buyer::class.java).single().second
val bobFuture = bobNode.smm.findStateMachines(BuyerAcceptor::class.java).single().second
// And off we go again.
net.runNetwork()
@ -489,25 +483,42 @@ class TwoPartyTradeFlowTests {
sellerNode: MockNetwork.MockNode,
buyerNode: MockNetwork.MockNode,
assetToSell: StateAndRef<OwnableState>): RunResult {
sellerNode.services.identityService.registerIdentity(buyerNode.info.legalIdentity)
buyerNode.services.identityService.registerIdentity(sellerNode.info.legalIdentity)
val buyerFlows: Observable<BuyerAcceptor> = buyerNode.registerInitiatedFlow(BuyerAcceptor::class.java)
val firstBuyerFiber = buyerFlows.toFuture().map { it.stateMachine }
val seller = SellerInitiator(buyerNode.info.legalIdentity, notaryNode.info, assetToSell, 1000.DOLLARS)
val sellerResult = sellerNode.services.startFlow(seller).resultFuture
return RunResult(firstBuyerFiber, sellerResult, seller.stateMachine.id)
}
@InitiatingFlow
class SellerRunnerFlow(val buyer: Party, val notary: NodeInfo) : FlowLogic<SignedTransaction>() {
class SellerInitiator(val buyer: Party,
val notary: NodeInfo,
val assetToSell: StateAndRef<OwnableState>,
val price: Amount<Currency>) : FlowLogic<SignedTransaction>() {
@Suspendable
override fun call(): SignedTransaction = subFlow(Seller(
override fun call(): SignedTransaction {
send(buyer, Pair(notary.notaryIdentity, price))
return subFlow(Seller(
buyer,
notary,
assetToSell,
1000.DOLLARS,
price,
serviceHub.legalIdentityKey))
}
}
sellerNode.services.identityService.registerIdentity(buyerNode.info.legalIdentity)
buyerNode.services.identityService.registerIdentity(sellerNode.info.legalIdentity)
val buyerFuture = buyerNode.initiateSingleShotFlow(SellerRunnerFlow::class) { otherParty ->
Buyer(otherParty, notaryNode.info.notaryIdentity, 1000.DOLLARS, CommercialPaper.State::class.java)
}.map { it.stateMachine }
val seller = SellerRunnerFlow(buyerNode.info.legalIdentity, notaryNode.info)
val sellerResultFuture = sellerNode.services.startFlow(seller).resultFuture
return RunResult(buyerFuture, sellerResultFuture, seller.stateMachine.id)
@InitiatedBy(SellerInitiator::class)
class BuyerAcceptor(val seller: Party) : FlowLogic<SignedTransaction>() {
@Suspendable
override fun call(): SignedTransaction {
val (notary, price) = receive<Pair<Party, Amount<Currency>>>(seller).unwrap {
require(serviceHub.networkMapCache.isNotary(it.first)) { "${it.first} is not a notary" }
it
}
return subFlow(Buyer(seller, notary, price, CommercialPaper.State::class.java))
}
}
private fun LedgerDSL<TestTransactionDSLInterpreter, TestLedgerDSLInterpreter>.runWithError(

10
node/src/test/kotlin/net/corda/node/services/MockServiceHubInternal.kt
View File

@ -3,11 +3,11 @@ package net.corda.node.services
import com.codahale.metrics.MetricRegistry
import net.corda.core.flows.FlowInitiator
import net.corda.core.flows.FlowLogic
import net.corda.core.identity.Party
import net.corda.core.node.NodeInfo
import net.corda.core.node.services.*
import net.corda.core.serialization.SerializeAsToken
import net.corda.core.transactions.SignedTransaction
import net.corda.node.internal.ServiceFlowInfo
import net.corda.node.internal.InitiatedFlowFactory
import net.corda.node.serialization.NodeClock
import net.corda.node.services.api.*
import net.corda.node.services.messaging.MessagingService
@ -67,11 +67,11 @@ open class MockServiceHubInternal(
override fun recordTransactions(txs: Iterable<SignedTransaction>) = recordTransactionsInternal(txStorageService, txs)
override fun <T : SerializeAsToken> cordaService(type: Class<T>): T = throw UnsupportedOperationException()
override fun <T> startFlow(logic: FlowLogic<T>, flowInitiator: FlowInitiator): FlowStateMachineImpl<T> {
return smm.executor.fetchFrom { smm.add(logic, flowInitiator) }
}
override fun registerServiceFlow(initiatingFlowClass: Class<out FlowLogic<*>>, serviceFlowFactory: (Party) -> FlowLogic<*>) = Unit
override fun getServiceFlowFactory(clientFlowClass: Class<out FlowLogic<*>>): ServiceFlowInfo? = null
override fun getFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>): InitiatedFlowFactory<*>? = null
}

4
node/src/test/kotlin/net/corda/node/services/events/NodeSchedulerServiceTest.kt
View File

@ -92,13 +92,13 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
}
scheduler = NodeSchedulerService(services, database, schedulerGatedExecutor)
smmExecutor = AffinityExecutor.ServiceAffinityExecutor("test", 1)
val mockSMM = StateMachineManager(services, listOf(services, scheduler), DBCheckpointStorage(), smmExecutor, database)
val mockSMM = StateMachineManager(services, DBCheckpointStorage(), smmExecutor, database)
mockSMM.changes.subscribe { change ->
if (change is StateMachineManager.Change.Removed && mockSMM.allStateMachines.isEmpty()) {
smmHasRemovedAllFlows.countDown()
}
}
mockSMM.start()
mockSMM.start(listOf(services, scheduler))
services.smm = mockSMM
scheduler.start()
}

11
node/src/test/kotlin/net/corda/node/services/persistence/DataVendingServiceTests.kt
View File

@ -6,9 +6,10 @@ import net.corda.core.contracts.Amount
import net.corda.core.contracts.Issued
import net.corda.core.contracts.TransactionType
import net.corda.core.contracts.USD
import net.corda.core.identity.Party
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.identity.Party
import net.corda.core.node.services.unconsumedStates
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.DUMMY_NOTARY
@ -86,16 +87,20 @@ class DataVendingServiceTests {
}
private fun MockNode.sendNotifyTx(tx: SignedTransaction, walletServiceNode: MockNode) {
walletServiceNode.registerServiceFlow(clientFlowClass = NotifyTxFlow::class, serviceFlowFactory = ::NotifyTransactionHandler)
walletServiceNode.registerInitiatedFlow(InitiateNotifyTxFlow::class.java)
services.startFlow(NotifyTxFlow(walletServiceNode.info.legalIdentity, tx))
network.runNetwork()
}
@InitiatingFlow
private class NotifyTxFlow(val otherParty: Party, val stx: SignedTransaction) : FlowLogic<Unit>() {
@Suspendable
override fun call() = send(otherParty, NotifyTxRequest(stx))
}
@InitiatedBy(NotifyTxFlow::class)
private class InitiateNotifyTxFlow(val otherParty: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() = subFlow(NotifyTransactionHandler(otherParty))
}
}

101
node/src/test/kotlin/net/corda/node/services/statemachine/FlowFrameworkTests.kt
View File

@ -7,13 +7,13 @@ import net.corda.contracts.asset.Cash
import net.corda.core.*
import net.corda.core.contracts.DOLLARS
import net.corda.core.contracts.DummyState
import net.corda.core.identity.Party
import net.corda.core.crypto.SecureHash
import net.corda.core.crypto.generateKeyPair
import net.corda.core.flows.FlowException
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.FlowSessionException
import net.corda.core.flows.InitiatingFlow
import net.corda.core.identity.Party
import net.corda.core.messaging.MessageRecipients
import net.corda.core.node.services.PartyInfo
import net.corda.core.node.services.ServiceInfo
@ -30,15 +30,19 @@ import net.corda.flows.CashIssueFlow
import net.corda.flows.CashPaymentFlow
import net.corda.flows.FinalityFlow
import net.corda.flows.NotaryFlow
import net.corda.node.internal.InitiatedFlowFactory
import net.corda.node.services.persistence.checkpoints
import net.corda.node.services.transactions.ValidatingNotaryService
import net.corda.node.utilities.transaction
import net.corda.testing.*
import net.corda.testing.expect
import net.corda.testing.expectEvents
import net.corda.testing.getTestX509Name
import net.corda.testing.node.InMemoryMessagingNetwork
import net.corda.testing.node.InMemoryMessagingNetwork.MessageTransfer
import net.corda.testing.node.InMemoryMessagingNetwork.ServicePeerAllocationStrategy.RoundRobin
import net.corda.testing.node.MockNetwork
import net.corda.testing.node.MockNetwork.MockNode
import net.corda.testing.sequence
import org.assertj.core.api.Assertions.assertThat
import org.assertj.core.api.Assertions.assertThatThrownBy
import org.assertj.core.api.AssertionsForClassTypes.assertThatExceptionOfType
@ -110,7 +114,7 @@ class FlowFrameworkTests {
@Test
fun `exception while fiber suspended`() {
node2.registerServiceFlow(ReceiveFlow::class) { SendFlow("Hello", it) }
node2.registerFlowFactory(ReceiveFlow::class) { SendFlow("Hello", it) }
val flow = ReceiveFlow(node2.info.legalIdentity)
val fiber = node1.services.startFlow(flow) as FlowStateMachineImpl
// Before the flow runs change the suspend action to throw an exception
@ -129,7 +133,7 @@ class FlowFrameworkTests {
@Test
fun `flow restarted just after receiving payload`() {
node2.registerServiceFlow(SendFlow::class) { ReceiveFlow(it).nonTerminating() }
node2.registerFlowFactory(SendFlow::class) { ReceiveFlow(it).nonTerminating() }
node1.services.startFlow(SendFlow("Hello", node2.info.legalIdentity))
// We push through just enough messages to get only the payload sent
@ -179,7 +183,7 @@ class FlowFrameworkTests {
@Test
fun `flow loaded from checkpoint will respond to messages from before start`() {
node1.registerServiceFlow(ReceiveFlow::class) { SendFlow("Hello", it) }
node1.registerFlowFactory(ReceiveFlow::class) { SendFlow("Hello", it) }
node2.services.startFlow(ReceiveFlow(node1.info.legalIdentity).nonTerminating()) // Prepare checkpointed receive flow
// Make sure the add() has finished initial processing.
node2.smm.executor.flush()
@ -198,7 +202,7 @@ class FlowFrameworkTests {
net.messagingNetwork.sentMessages.toSessionTransfers().filter { it.isPayloadTransfer }.forEach { sentCount++ }
val node3 = net.createNode(node1.info.address)
val secondFlow = node3.initiateSingleShotFlow(PingPongFlow::class) { PingPongFlow(it, payload2) }
val secondFlow = node3.registerFlowFactory(PingPongFlow::class) { PingPongFlow(it, payload2) }
net.runNetwork()
// Kick off first send and receive
@ -243,8 +247,8 @@ class FlowFrameworkTests {
fun `sending to multiple parties`() {
val node3 = net.createNode(node1.info.address)
net.runNetwork()
node2.registerServiceFlow(SendFlow::class) { ReceiveFlow(it).nonTerminating() }
node3.registerServiceFlow(SendFlow::class) { ReceiveFlow(it).nonTerminating() }
node2.registerFlowFactory(SendFlow::class) { ReceiveFlow(it).nonTerminating() }
node3.registerFlowFactory(SendFlow::class) { ReceiveFlow(it).nonTerminating() }
val payload = "Hello World"
node1.services.startFlow(SendFlow(payload, node2.info.legalIdentity, node3.info.legalIdentity))
net.runNetwork()
@ -277,8 +281,8 @@ class FlowFrameworkTests {
net.runNetwork()
val node2Payload = "Test 1"
val node3Payload = "Test 2"
node2.registerServiceFlow(ReceiveFlow::class) { SendFlow(node2Payload, it) }
node3.registerServiceFlow(ReceiveFlow::class) { SendFlow(node3Payload, it) }
node2.registerFlowFactory(ReceiveFlow::class) { SendFlow(node2Payload, it) }
node3.registerFlowFactory(ReceiveFlow::class) { SendFlow(node3Payload, it) }
val multiReceiveFlow = ReceiveFlow(node2.info.legalIdentity, node3.info.legalIdentity).nonTerminating()
node1.services.startFlow(multiReceiveFlow)
node1.acceptableLiveFiberCountOnStop = 1
@ -303,7 +307,7 @@ class FlowFrameworkTests {
@Test
fun `both sides do a send as their first IO request`() {
node2.registerServiceFlow(PingPongFlow::class) { PingPongFlow(it, 20L) }
node2.registerFlowFactory(PingPongFlow::class) { PingPongFlow(it, 20L) }
node1.services.startFlow(PingPongFlow(node2.info.legalIdentity, 10L))
net.runNetwork()
@ -339,7 +343,7 @@ class FlowFrameworkTests {
sessionTransfers.expectEvents(isStrict = false) {
sequence(
// First Pay
expect(match = { it.message is SessionInit && it.message.clientFlowClass == NotaryFlow.Client::class.java }) {
expect(match = { it.message is SessionInit && it.message.initiatingFlowClass == NotaryFlow.Client::class.java }) {
it.message as SessionInit
assertEquals(node1.id, it.from)
assertEquals(notary1Address, it.to)
@ -349,7 +353,7 @@ class FlowFrameworkTests {
assertEquals(notary1.id, it.from)
},
// Second pay
expect(match = { it.message is SessionInit && it.message.clientFlowClass == NotaryFlow.Client::class.java }) {
expect(match = { it.message is SessionInit && it.message.initiatingFlowClass == NotaryFlow.Client::class.java }) {
it.message as SessionInit
assertEquals(node1.id, it.from)
assertEquals(notary1Address, it.to)
@ -359,7 +363,7 @@ class FlowFrameworkTests {
assertEquals(notary2.id, it.from)
},
// Third pay
expect(match = { it.message is SessionInit && it.message.clientFlowClass == NotaryFlow.Client::class.java }) {
expect(match = { it.message is SessionInit && it.message.initiatingFlowClass == NotaryFlow.Client::class.java }) {
it.message as SessionInit
assertEquals(node1.id, it.from)
assertEquals(notary1Address, it.to)
@ -374,7 +378,7 @@ class FlowFrameworkTests {
@Test
fun `other side ends before doing expected send`() {
node2.registerServiceFlow(ReceiveFlow::class) { NoOpFlow() }
node2.registerFlowFactory(ReceiveFlow::class) { NoOpFlow() }
val resultFuture = node1.services.startFlow(ReceiveFlow(node2.info.legalIdentity)).resultFuture
net.runNetwork()
assertThatExceptionOfType(FlowSessionException::class.java).isThrownBy {
@ -384,7 +388,7 @@ class FlowFrameworkTests {
@Test
fun `non-FlowException thrown on other side`() {
val erroringFlowFuture = node2.initiateSingleShotFlow(ReceiveFlow::class) {
val erroringFlowFuture = node2.registerFlowFactory(ReceiveFlow::class) {
ExceptionFlow { Exception("evil bug!") }
}
val erroringFlowSteps = erroringFlowFuture.flatMap { it.progressSteps }
@ -418,7 +422,7 @@ class FlowFrameworkTests {
@Test
fun `FlowException thrown on other side`() {
val erroringFlow = node2.initiateSingleShotFlow(ReceiveFlow::class) {
val erroringFlow = node2.registerFlowFactory(ReceiveFlow::class) {
ExceptionFlow { MyFlowException("Nothing useful") }
}
val erroringFlowSteps = erroringFlow.flatMap { it.progressSteps }
@ -456,8 +460,8 @@ class FlowFrameworkTests {
val node3 = net.createNode(node1.info.address)
net.runNetwork()
node3.initiateSingleShotFlow(ReceiveFlow::class) { ExceptionFlow { MyFlowException("Chain") } }
node2.initiateSingleShotFlow(ReceiveFlow::class) { ReceiveFlow(node3.info.legalIdentity) }
node3.registerFlowFactory(ReceiveFlow::class) { ExceptionFlow { MyFlowException("Chain") } }
node2.registerFlowFactory(ReceiveFlow::class) { ReceiveFlow(node3.info.legalIdentity) }
val receivingFiber = node1.services.startFlow(ReceiveFlow(node2.info.legalIdentity))
net.runNetwork()
assertThatExceptionOfType(MyFlowException::class.java)
@ -473,9 +477,9 @@ class FlowFrameworkTests {
// Node 2 will send its payload and then block waiting for the receive from node 1. Meanwhile node 1 will move
// onto node 3 which will throw the exception
val node2Fiber = node2
.initiateSingleShotFlow(ReceiveFlow::class) { SendAndReceiveFlow(it, "Hello") }
.registerFlowFactory(ReceiveFlow::class) { SendAndReceiveFlow(it, "Hello") }
.map { it.stateMachine }
node3.initiateSingleShotFlow(ReceiveFlow::class) { ExceptionFlow { MyFlowException("Nothing useful") } }
node3.registerFlowFactory(ReceiveFlow::class) { ExceptionFlow { MyFlowException("Nothing useful") } }
val node1Fiber = node1.services.startFlow(ReceiveFlow(node2.info.legalIdentity, node3.info.legalIdentity)) as FlowStateMachineImpl
net.runNetwork()
@ -528,7 +532,7 @@ class FlowFrameworkTests {
}
}
node2.registerServiceFlow(AskForExceptionFlow::class) { ConditionalExceptionFlow(it, "Hello") }
node2.registerFlowFactory(AskForExceptionFlow::class) { ConditionalExceptionFlow(it, "Hello") }
val resultFuture = node1.services.startFlow(RetryOnExceptionFlow(node2.info.legalIdentity)).resultFuture
net.runNetwork()
assertThat(resultFuture.getOrThrow()).isEqualTo("Hello")
@ -536,7 +540,7 @@ class FlowFrameworkTests {
@Test
fun `serialisation issue in counterparty`() {
node2.registerServiceFlow(ReceiveFlow::class) { SendFlow(NonSerialisableData(1), it) }
node2.registerFlowFactory(ReceiveFlow::class) { SendFlow(NonSerialisableData(1), it) }
val result = node1.services.startFlow(ReceiveFlow(node2.info.legalIdentity)).resultFuture
net.runNetwork()
assertThatExceptionOfType(FlowSessionException::class.java).isThrownBy {
@ -546,7 +550,7 @@ class FlowFrameworkTests {
@Test
fun `FlowException has non-serialisable object`() {
node2.initiateSingleShotFlow(ReceiveFlow::class) {
node2.registerFlowFactory(ReceiveFlow::class) {
ExceptionFlow { NonSerialisableFlowException(NonSerialisableData(1)) }
}
val result = node1.services.startFlow(ReceiveFlow(node2.info.legalIdentity)).resultFuture
@ -562,9 +566,9 @@ class FlowFrameworkTests {
ptx.addOutputState(DummyState())
val stx = node1.services.signInitialTransaction(ptx)
val committerFiber = node1
.initiateSingleShotFlow(WaitingFlows.Waiter::class) { WaitingFlows.Committer(it) }
.map { it.stateMachine }
val committerFiber = node1.registerFlowFactory(WaitingFlows.Waiter::class) {
WaitingFlows.Committer(it)
}.map { it.stateMachine }
val waiterStx = node2.services.startFlow(WaitingFlows.Waiter(stx, node1.info.legalIdentity)).resultFuture
net.runNetwork()
assertThat(waiterStx.getOrThrow()).isEqualTo(committerFiber.getOrThrow().resultFuture.getOrThrow())
@ -576,7 +580,7 @@ class FlowFrameworkTests {
ptx.addOutputState(DummyState())
val stx = node1.services.signInitialTransaction(ptx)
node1.registerServiceFlow(WaitingFlows.Waiter::class) {
node1.registerFlowFactory(WaitingFlows.Waiter::class) {
WaitingFlows.Committer(it) { throw Exception("Error") }
}
val waiter = node2.services.startFlow(WaitingFlows.Waiter(stx, node1.info.legalIdentity)).resultFuture
@ -594,13 +598,22 @@ class FlowFrameworkTests {
}
@Test
fun `custom client flow`() {
val receiveFlowFuture = node2.initiateSingleShotFlow(SendFlow::class) { ReceiveFlow(it) }
fun `customised client flow`() {
val receiveFlowFuture = node2.registerFlowFactory(SendFlow::class) { ReceiveFlow(it) }
node1.services.startFlow(CustomSendFlow("Hello", node2.info.legalIdentity)).resultFuture
net.runNetwork()
assertThat(receiveFlowFuture.getOrThrow().receivedPayloads).containsOnly("Hello")
}
@Test
fun `customised client flow which has annotated @InitiatingFlow again`() {
val result = node1.services.startFlow(IncorrectCustomSendFlow("Hello", node2.info.legalIdentity)).resultFuture
net.runNetwork()
assertThatExceptionOfType(IllegalArgumentException::class.java).isThrownBy {
result.getOrThrow()
}.withMessageContaining(InitiatingFlow::class.java.simpleName)
}
@Test
fun `upgraded flow`() {
node1.services.startFlow(UpgradedFlow(node2.info.legalIdentity))
@ -612,7 +625,11 @@ class FlowFrameworkTests {
@Test
fun `unsupported new flow version`() {
node2.registerServiceFlow(UpgradedFlow::class, flowVersion = 1) { SendFlow("Hello", it) }
node2.registerFlowFactory(
UpgradedFlow::class.java,
InitiatedFlowFactory.CorDapp(version = 1, factory = ::DoubleInlinedSubFlow),
DoubleInlinedSubFlow::class.java,
track = false)
val result = node1.services.startFlow(UpgradedFlow(node2.info.legalIdentity)).resultFuture
net.runNetwork()
assertThatExceptionOfType(FlowSessionException::class.java).isThrownBy {
@ -622,7 +639,7 @@ class FlowFrameworkTests {
@Test
fun `single inlined sub-flow`() {
node2.registerServiceFlow(SendAndReceiveFlow::class) { SingleInlinedSubFlow(it) }
node2.registerFlowFactory(SendAndReceiveFlow::class, ::SingleInlinedSubFlow)
val result = node1.services.startFlow(SendAndReceiveFlow(node2.info.legalIdentity, "Hello")).resultFuture
net.runNetwork()
assertThat(result.getOrThrow()).isEqualTo("HelloHello")
@ -630,7 +647,7 @@ class FlowFrameworkTests {
@Test
fun `double inlined sub-flow`() {
node2.registerServiceFlow(SendAndReceiveFlow::class) { DoubleInlinedSubFlow(it) }
node2.registerFlowFactory(SendAndReceiveFlow::class, ::DoubleInlinedSubFlow)
val result = node1.services.startFlow(SendAndReceiveFlow(node2.info.legalIdentity, "Hello")).resultFuture
net.runNetwork()
assertThat(result.getOrThrow()).isEqualTo("HelloHello")
@ -654,6 +671,18 @@ class FlowFrameworkTests {
return smm.findStateMachines(P::class.java).single()
}
private inline fun <reified P : FlowLogic<*>> MockNode.registerFlowFactory(
initiatingFlowClass: KClass<out FlowLogic<*>>,
noinline flowFactory: (Party) -> P): ListenableFuture<P>
{
val observable = registerFlowFactory(initiatingFlowClass.java, object : InitiatedFlowFactory<P> {
override fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): P {
return flowFactory(otherParty)
}
}, P::class.java, track = true)
return observable.toFuture()
}
private fun sessionInit(clientFlowClass: KClass<out FlowLogic<*>>, flowVersion: Int = 1, payload: Any? = null): SessionInit {
return SessionInit(0, clientFlowClass.java, flowVersion, payload)
}
@ -730,8 +759,12 @@ class FlowFrameworkTests {
}
private interface CustomInterface
private class CustomSendFlow(payload: String, otherParty: Party) : CustomInterface, SendFlow(payload, otherParty)
@InitiatingFlow
private class IncorrectCustomSendFlow(payload: String, otherParty: Party) : CustomInterface, SendFlow(payload, otherParty)
@InitiatingFlow
private class ReceiveFlow(vararg val otherParties: Party) : FlowLogic<Unit>() {
object START_STEP : ProgressTracker.Step("Starting")
@ -852,7 +885,7 @@ class FlowFrameworkTests {
}
private data class NonSerialisableData(val a: Int)
private class NonSerialisableFlowException(val data: NonSerialisableData) : FlowException()
private class NonSerialisableFlowException(@Suppress("unused") val data: NonSerialisableData) : FlowException()
//endregion Helpers
}

3
samples/bank-of-corda-demo/src/main/kotlin/net/corda/bank/plugin/BankOfCordaPlugin.kt
View File

@ -1,13 +1,10 @@
package net.corda.bank.plugin
import net.corda.bank.api.BankOfCordaWebApi
import net.corda.core.identity.Party
import net.corda.core.node.CordaPluginRegistry
import net.corda.flows.IssuerFlow
import java.util.function.Function
class BankOfCordaPlugin : CordaPluginRegistry() {
// A list of classes that expose web APIs.
override val webApis = listOf(Function(::BankOfCordaWebApi))
override val servicePlugins = listOf(Function(IssuerFlow.Issuer::Service))
}

14
samples/irs-demo/src/integration-test/kotlin/net/corda/irs/IRSDemoTest.kt
View File

@ -33,7 +33,11 @@ class IRSDemoTest : IntegrationTestCategory {
@Test
fun `runs IRS demo`() {
driver(useTestClock = true, isDebug = true) {
driver(
useTestClock = true,
isDebug = true,
systemProperties = mapOf("net.corda.node.cordapp.scan.package" to "net.corda.irs"))
{
val (controller, nodeA, nodeB) = Futures.allAsList(
startNode(DUMMY_NOTARY.name, setOf(ServiceInfo(SimpleNotaryService.type), ServiceInfo(NodeInterestRates.type))),
startNode(DUMMY_BANK_A.name, rpcUsers = listOf(rpcUser)),
@ -83,18 +87,22 @@ class IRSDemoTest : IntegrationTestCategory {
}
private fun runTrade(nodeAddr: HostAndPort) {
val fileContents = IOUtils.toString(Thread.currentThread().contextClassLoader.getResourceAsStream("net/corda/irs/simulation/example-irs-trade.json"), Charsets.UTF_8.name())
val fileContents = loadResourceFile("net/corda/irs/simulation/example-irs-trade.json")
val tradeFile = fileContents.replace("tradeXXX", "trade1")
val url = URL("http://$nodeAddr/api/irs/deals")
assertThat(postJson(url, tradeFile)).isTrue()
}
private fun runUploadRates(host: HostAndPort) {
val fileContents = IOUtils.toString(Thread.currentThread().contextClassLoader.getResourceAsStream("net/corda/irs/simulation/example.rates.txt"), Charsets.UTF_8.name())
val fileContents = loadResourceFile("net/corda/irs/simulation/example.rates.txt")
val url = URL("http://$host/upload/interest-rates")
assertThat(uploadFile(url, fileContents)).isTrue()
}
private fun loadResourceFile(filename: String): String {
return IOUtils.toString(Thread.currentThread().contextClassLoader.getResourceAsStream(filename), Charsets.UTF_8.name())
}
private fun getTradeCount(nodeAddr: HostAndPort): Int {
val api = HttpApi.fromHostAndPort(nodeAddr, "api/irs")
val deals = api.getJson<Array<*>>("deals")

87
samples/irs-demo/src/main/kotlin/net/corda/irs/api/NodeInterestRates.kt
View File

@ -6,15 +6,15 @@ import net.corda.core.contracts.*
import net.corda.core.crypto.DigitalSignature
import net.corda.core.crypto.MerkleTreeException
import net.corda.core.crypto.keys
import net.corda.core.crypto.sign
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.identity.Party
import net.corda.core.math.CubicSplineInterpolator
import net.corda.core.math.Interpolator
import net.corda.core.math.InterpolatorFactory
import net.corda.core.node.CordaPluginRegistry
import net.corda.core.node.PluginServiceHub
import net.corda.core.node.ServiceHub
import net.corda.core.node.services.CordaService
import net.corda.core.node.services.ServiceType
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.transactions.FilteredTransaction
@ -30,14 +30,10 @@ import org.jetbrains.exposed.sql.ResultRow
import org.jetbrains.exposed.sql.statements.InsertStatement
import java.io.InputStream
import java.math.BigDecimal
import java.security.KeyPair
import java.time.Clock
import java.security.PublicKey
import java.time.Duration
import java.time.Instant
import java.time.LocalDate
import java.util.*
import java.util.function.Function
import javax.annotation.concurrent.ThreadSafe
import kotlin.collections.component1
import kotlin.collections.component2
@ -55,83 +51,50 @@ import kotlin.collections.set
object NodeInterestRates {
val type = ServiceType.corda.getSubType("interest_rates")
/**
* Register the flow that is used with the Fixing integration tests.
*/
class Plugin : CordaPluginRegistry() {
override val servicePlugins = listOf(Function(::Service))
}
/**
* The Service that wraps [Oracle] and handles messages/network interaction/request scrubbing.
*/
// DOCSTART 2
class Service(val services: PluginServiceHub) : AcceptsFileUpload, SingletonSerializeAsToken() {
val oracle: Oracle by lazy {
val myNodeInfo = services.myInfo
val myIdentity = myNodeInfo.serviceIdentities(type).first()
val mySigningKey = myIdentity.owningKey.keys.first { services.keyManagementService.keys.contains(it) }
Oracle(myIdentity, mySigningKey, services)
}
init {
// Note: access to the singleton oracle property is via the registered SingletonSerializeAsToken Service.
// Otherwise the Kryo serialisation of the call stack in the Quasar Fiber extends to include
// the framework Oracle and the flow will crash.
services.registerServiceFlow(RatesFixFlow.FixSignFlow::class.java) { FixSignHandler(it, this) }
services.registerServiceFlow(RatesFixFlow.FixQueryFlow::class.java) { FixQueryHandler(it, this) }
}
private class FixSignHandler(val otherParty: Party, val service: Service) : FlowLogic<Unit>() {
@InitiatedBy(RatesFixFlow.FixSignFlow::class)
class FixSignHandler(val otherParty: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() {
val request = receive<RatesFixFlow.SignRequest>(otherParty).unwrap { it }
send(otherParty, service.oracle.sign(request.ftx))
send(otherParty, serviceHub.cordaService(Oracle::class.java).sign(request.ftx))
}
}
private class FixQueryHandler(val otherParty: Party, val service: Service) : FlowLogic<Unit>() {
companion object {
@InitiatedBy(RatesFixFlow.FixQueryFlow::class)
class FixQueryHandler(val otherParty: Party) : FlowLogic<Unit>() {
object RECEIVED : ProgressTracker.Step("Received fix request")
object SENDING : ProgressTracker.Step("Sending fix response")
}
override val progressTracker = ProgressTracker(RECEIVED, SENDING)
init {
progressTracker.currentStep = RECEIVED
}
@Suspendable
override fun call(): Unit {
val request = receive<RatesFixFlow.QueryRequest>(otherParty).unwrap { it }
val answers = service.oracle.query(request.queries, request.deadline)
progressTracker.currentStep = RECEIVED
val answers = serviceHub.cordaService(Oracle::class.java).query(request.queries, request.deadline)
progressTracker.currentStep = SENDING
send(otherParty, answers)
}
}
// DOCEND 2
// File upload support
override val dataTypePrefix = "interest-rates"
override val acceptableFileExtensions = listOf(".rates", ".txt")
override fun upload(file: InputStream): String {
val fixes = parseFile(file.bufferedReader().readText())
oracle.knownFixes = fixes
val msg = "Interest rates oracle accepted ${fixes.size} new interest rate fixes"
println(msg)
return msg
}
}
/**
* An implementation of an interest rate fix oracle which is given data in a simple string format.
*
* The oracle will try to interpolate the missing value of a tenor for the given fix name and date.
*/
@ThreadSafe
class Oracle(val identity: Party, private val signingKey: PublicKey, val services: ServiceHub) {
// DOCSTART 3
@CordaService
class Oracle(val identity: Party, private val signingKey: PublicKey, val services: ServiceHub) : AcceptsFileUpload, SingletonSerializeAsToken() {
constructor(services: PluginServiceHub) : this(
services.myInfo.serviceIdentities(type).first(),
services.myInfo.serviceIdentities(type).first().owningKey.keys.first { services.keyManagementService.keys.contains(it) },
services
)
// DOCEND 3
private object Table : JDBCHashedTable("demo_interest_rate_fixes") {
val name = varchar("index_name", length = 255)
val forDay = localDate("for_day")
@ -175,7 +138,7 @@ object NodeInterestRates {
/**
* This method will now wait until the given deadline if the fix for the given [FixOf] is not immediately
* available. To implement this, [readWithDeadline] will loop if the deadline is not reached and we throw
* available. To implement this, [FiberBox.readWithDeadline] will loop if the deadline is not reached and we throw
* [UnknownFix] as it implements [RetryableException] which has special meaning to this function.
*/
@Suspendable
@ -231,6 +194,16 @@ object NodeInterestRates {
return DigitalSignature.LegallyIdentifiable(identity, signature.bytes)
}
// DOCEND 1
// File upload support
override val dataTypePrefix = "interest-rates"
override val acceptableFileExtensions = listOf(".rates", ".txt")
override fun upload(file: InputStream): String {
val fixes = parseFile(file.bufferedReader().readText())
knownFixes = fixes
return "Interest rates oracle accepted ${fixes.size} new interest rate fixes"
}
}
// TODO: can we split into two? Fix not available (retryable/transient) and unknown (permanent)

21
samples/irs-demo/src/main/kotlin/net/corda/irs/flows/AutoOfferFlow.kt
View File

@ -3,19 +3,17 @@ package net.corda.irs.flows
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.contracts.DealState
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StartableByRPC
import net.corda.core.identity.AbstractParty
import net.corda.core.node.CordaPluginRegistry
import net.corda.core.node.PluginServiceHub
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.identity.Party
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.ProgressTracker
import net.corda.flows.TwoPartyDealFlow
import net.corda.flows.TwoPartyDealFlow.Acceptor
import net.corda.flows.TwoPartyDealFlow.AutoOffer
import net.corda.flows.TwoPartyDealFlow.Instigator
import java.util.function.Function
/**
* This whole class is really part of a demo just to initiate the agreement of a deal with a simple
@ -25,18 +23,6 @@ import java.util.function.Function
* or the flow would have to reach out to external systems (or users) to verify the deals.
*/
object AutoOfferFlow {
class Plugin : CordaPluginRegistry() {
override val servicePlugins = listOf(Function(::Service))
}
class Service(services: PluginServiceHub) : SingletonSerializeAsToken() {
init {
services.registerServiceFlow(Requester::class.java) { Acceptor(it) }
}
}
@InitiatingFlow
@StartableByRPC
class Requester(val dealToBeOffered: DealState) : FlowLogic<SignedTransaction>() {
@ -81,4 +67,7 @@ object AutoOfferFlow {
return parties.filter { serviceHub.myInfo.legalIdentity != it }
}
}
@InitiatedBy(Requester::class)
class AutoOfferAcceptor(otherParty: Party) : Acceptor(otherParty)
}

14
samples/irs-demo/src/main/kotlin/net/corda/irs/flows/FixingFlow.kt
View File

@ -5,11 +5,11 @@ import net.corda.core.TransientProperty
import net.corda.core.contracts.*
import net.corda.core.crypto.toBase58String
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.SchedulableFlow
import net.corda.core.identity.Party
import net.corda.core.node.NodeInfo
import net.corda.core.node.PluginServiceHub
import net.corda.core.node.services.ServiceType
import net.corda.core.seconds
import net.corda.core.serialization.CordaSerializable
@ -22,13 +22,6 @@ import java.math.BigDecimal
import java.security.PublicKey
object FixingFlow {
class Service(services: PluginServiceHub) {
init {
services.registerServiceFlow(FixingRoleDecider::class.java) { Fixer(it) }
}
}
/**
* One side of the fixing flow for an interest rate swap, but could easily be generalised further.
*
@ -36,8 +29,9 @@ object FixingFlow {
* of the flow that is run by the party with the fixed leg of swap deal, which is the basis for deciding
* who does what in the flow.
*/
class Fixer(override val otherParty: Party,
override val progressTracker: ProgressTracker = TwoPartyDealFlow.Secondary.tracker()) : TwoPartyDealFlow.Secondary<FixingSession>() {
@InitiatedBy(FixingRoleDecider::class)
class Fixer(override val otherParty: Party) : TwoPartyDealFlow.Secondary<FixingSession>() {
override val progressTracker: ProgressTracker = TwoPartyDealFlow.Secondary.tracker()
private lateinit var txState: TransactionState<*>
private lateinit var deal: FixableDealState

15
samples/irs-demo/src/main/kotlin/net/corda/irs/flows/UpdateBusinessDayFlow.kt
View File

@ -2,19 +2,17 @@ package net.corda.irs.flows
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StartableByRPC
import net.corda.core.identity.Party
import net.corda.core.node.CordaPluginRegistry
import net.corda.core.node.NodeInfo
import net.corda.core.node.PluginServiceHub
import net.corda.core.serialization.CordaSerializable
import net.corda.core.utilities.ProgressTracker
import net.corda.core.utilities.unwrap
import net.corda.node.utilities.TestClock
import net.corda.testing.node.MockNetworkMapCache
import java.time.LocalDate
import java.util.function.Function
/**
* This is a less temporary, demo-oriented way of initiating processing of temporal events.
@ -26,16 +24,7 @@ object UpdateBusinessDayFlow {
@CordaSerializable
data class UpdateBusinessDayMessage(val date: LocalDate)
class Plugin : CordaPluginRegistry() {
override val servicePlugins = listOf(Function(::Service))
}
class Service(services: PluginServiceHub) {
init {
services.registerServiceFlow(Broadcast::class.java, ::UpdateBusinessDayHandler)
}
}
@InitiatedBy(Broadcast::class)
private class UpdateBusinessDayHandler(val otherParty: Party) : FlowLogic<Unit>() {
override fun call() {
val message = receive<UpdateBusinessDayMessage>(otherParty).unwrap { it }

2
samples/irs-demo/src/main/kotlin/net/corda/irs/plugin/IRSPlugin.kt
View File

@ -2,7 +2,6 @@ package net.corda.irs.plugin
import net.corda.core.node.CordaPluginRegistry
import net.corda.irs.api.InterestRateSwapAPI
import net.corda.irs.flows.FixingFlow
import java.util.function.Function
class IRSPlugin : CordaPluginRegistry() {
@ -10,5 +9,4 @@ class IRSPlugin : CordaPluginRegistry() {
override val staticServeDirs: Map<String, String> = mapOf(
"irsdemo" to javaClass.classLoader.getResource("irsweb").toExternalForm()
)
override val servicePlugins = listOf(Function(FixingFlow::Service))
}

23
samples/irs-demo/src/main/kotlin/net/corda/irs/simulation/IRSSimulation.kt
View File

@ -7,27 +7,26 @@ import com.google.common.util.concurrent.FutureCallback
import com.google.common.util.concurrent.Futures
import com.google.common.util.concurrent.ListenableFuture
import com.google.common.util.concurrent.SettableFuture
import net.corda.core.RunOnCallerThread
import net.corda.core.*
import net.corda.core.contracts.StateAndRef
import net.corda.core.contracts.UniqueIdentifier
import net.corda.core.flatMap
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.FlowStateMachine
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.identity.Party
import net.corda.core.map
import net.corda.core.node.services.linearHeadsOfType
import net.corda.core.success
import net.corda.core.transactions.SignedTransaction
import net.corda.flows.TwoPartyDealFlow.Acceptor
import net.corda.flows.TwoPartyDealFlow.AutoOffer
import net.corda.flows.TwoPartyDealFlow.Instigator
import net.corda.irs.contract.InterestRateSwap
import net.corda.irs.flows.FixingFlow
import net.corda.jackson.JacksonSupport
import net.corda.node.services.identity.InMemoryIdentityService
import net.corda.node.utilities.transaction
import net.corda.testing.initiateSingleShotFlow
import net.corda.testing.node.InMemoryMessagingNetwork
import rx.Observable
import java.security.PublicKey
import java.time.LocalDate
import java.util.*
@ -126,6 +125,9 @@ class IRSSimulation(networkSendManuallyPumped: Boolean, runAsync: Boolean, laten
irs.fixedLeg.fixedRatePayer = node1.info.legalIdentity
irs.floatingLeg.floatingRatePayer = node2.info.legalIdentity
node1.registerInitiatedFlow(FixingFlow.Fixer::class.java)
node2.registerInitiatedFlow(FixingFlow.Fixer::class.java)
@InitiatingFlow
class StartDealFlow(val otherParty: Party,
val payload: AutoOffer,
@ -134,8 +136,13 @@ class IRSSimulation(networkSendManuallyPumped: Boolean, runAsync: Boolean, laten
override fun call(): SignedTransaction = subFlow(Instigator(otherParty, payload, myKey))
}
@InitiatedBy(StartDealFlow::class)
class AcceptDealFlow(otherParty: Party) : Acceptor(otherParty)
val acceptDealFlows: Observable<AcceptDealFlow> = node2.registerInitiatedFlow(AcceptDealFlow::class.java)
@Suppress("UNCHECKED_CAST")
val acceptorTx = node2.initiateSingleShotFlow(StartDealFlow::class) { Acceptor(it) }.flatMap {
val acceptorTxFuture = acceptDealFlows.toFuture().flatMap {
(it.stateMachine as FlowStateMachine<SignedTransaction>).resultFuture
}
@ -146,9 +153,9 @@ class IRSSimulation(networkSendManuallyPumped: Boolean, runAsync: Boolean, laten
node2.info.legalIdentity,
AutoOffer(notary.info.notaryIdentity, irs),
node1.services.legalIdentityKey)
val instigatorTx = node1.services.startFlow(instigator).resultFuture
val instigatorTxFuture = node1.services.startFlow(instigator).resultFuture
return Futures.allAsList(instigatorTx, acceptorTx).flatMap { instigatorTx }
return Futures.allAsList(instigatorTxFuture, acceptorTxFuture).flatMap { instigatorTxFuture }
}
override fun iterate(): InMemoryMessagingNetwork.MessageTransfer? {

4
samples/irs-demo/src/main/kotlin/net/corda/irs/simulation/Simulation.kt
View File

@ -126,9 +126,11 @@ abstract class Simulation(val networkSendManuallyPumped: Boolean,
return object : SimulatedNode(cfg, network, networkMapAddr, advertisedServices, id, overrideServices, entropyRoot) {
override fun start(): MockNetwork.MockNode {
super.start()
registerInitiatedFlow(NodeInterestRates.FixQueryHandler::class.java)
registerInitiatedFlow(NodeInterestRates.FixSignHandler::class.java)
javaClass.classLoader.getResourceAsStream("net/corda/irs/simulation/example.rates.txt").use {
database.transaction {
findService<NodeInterestRates.Service>().upload(it)
installCordaService(NodeInterestRates.Oracle::class.java).upload(it)
}
}
return this

3
samples/irs-demo/src/main/resources/META-INF/services/net.corda.core.node.CordaPluginRegistry
View File

@ -1,5 +1,2 @@
# Register a ServiceLoader service extending from net.corda.core.node.CordaPluginRegistry
net.corda.irs.plugin.IRSPlugin
net.corda.irs.api.NodeInterestRates$Plugin
net.corda.irs.flows.AutoOfferFlow$Plugin
net.corda.irs.flows.UpdateBusinessDayFlow$Plugin

10
samples/irs-demo/src/test/kotlin/net/corda/irs/testing/NodeInterestRatesTest.kt
View File

@ -210,8 +210,10 @@ class NodeInterestRatesTest {
val net = MockNetwork()
val n1 = net.createNotaryNode()
val n2 = net.createNode(n1.info.address, advertisedServices = ServiceInfo(NodeInterestRates.type))
n2.registerInitiatedFlow(NodeInterestRates.FixQueryHandler::class.java)
n2.registerInitiatedFlow(NodeInterestRates.FixSignHandler::class.java)
n2.database.transaction {
n2.findService<NodeInterestRates.Service>().oracle.knownFixes = TEST_DATA
n2.installCordaService(NodeInterestRates.Oracle::class.java).knownFixes = TEST_DATA
}
val tx = TransactionType.General.Builder(null)
val fixOf = NodeInterestRates.parseFixOf("LIBOR 2016-03-16 1M")
@ -233,7 +235,8 @@ class NodeInterestRatesTest {
fixOf: FixOf,
expectedRate: BigDecimal,
rateTolerance: BigDecimal,
progressTracker: ProgressTracker = RatesFixFlow.tracker(fixOf.name)) : RatesFixFlow(tx, oracle, fixOf, expectedRate, rateTolerance, progressTracker) {
progressTracker: ProgressTracker = RatesFixFlow.tracker(fixOf.name))
: RatesFixFlow(tx, oracle, fixOf, expectedRate, rateTolerance, progressTracker) {
override fun filtering(elem: Any): Boolean {
return when (elem) {
is Command -> oracle.owningKey in elem.signers && elem.value is Fix
@ -242,5 +245,6 @@ class NodeInterestRatesTest {
}
}
private fun makeTX() = TransactionType.General.Builder(DUMMY_NOTARY).withItems(1000.DOLLARS.CASH `issued by` DUMMY_CASH_ISSUER `owned by` ALICE `with notary` DUMMY_NOTARY)
private fun makeTX() = TransactionType.General.Builder(DUMMY_NOTARY).withItems(
1000.DOLLARS.CASH `issued by` DUMMY_CASH_ISSUER `owned by` ALICE `with notary` DUMMY_NOTARY)
}

7
samples/simm-valuation-demo/src/integration-test/kotlin/net/corda/vega/SimmValuationTest.kt
View File

@ -32,7 +32,7 @@ class SimmValuationTest : IntegrationTestCategory {
@Test
fun `runs SIMM valuation demo`() {
driver(isDebug = true) {
driver(isDebug = true, systemProperties = mapOf("net.corda.node.cordapp.scan.package" to "net.corda.vega")) {
startNode(DUMMY_NOTARY.name, setOf(ServiceInfo(SimpleNotaryService.type))).getOrThrow()
val (nodeA, nodeB) = Futures.allAsList(startNode(nodeALegalName), startNode(nodeBLegalName)).getOrThrow()
val (nodeAApi, nodeBApi) = Futures.allAsList(startWebserver(nodeA), startWebserver(nodeB))
@ -50,8 +50,9 @@ class SimmValuationTest : IntegrationTestCategory {
}
}
private fun getPartyWithName(partyApi: HttpApi, counterparty: X500Name): PortfolioApi.ApiParty =
getAvailablePartiesFor(partyApi).counterparties.single { it.text == counterparty }
private fun getPartyWithName(partyApi: HttpApi, counterparty: X500Name): PortfolioApi.ApiParty {
return getAvailablePartiesFor(partyApi).counterparties.single { it.text == counterparty }
}
private fun getAvailablePartiesFor(partyApi: HttpApi): PortfolioApi.AvailableParties {
return partyApi.getJson<PortfolioApi.AvailableParties>("whoami")

9
samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/flows/IRSTradeFlow.kt
View File

@ -2,10 +2,10 @@ package net.corda.vega.flows
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StartableByRPC
import net.corda.core.identity.Party
import net.corda.core.node.PluginServiceHub
import net.corda.core.serialization.CordaSerializable
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.unwrap
@ -15,12 +15,6 @@ import net.corda.vega.contracts.OGTrade
import net.corda.vega.contracts.SwapData
object IRSTradeFlow {
class Service(services: PluginServiceHub) {
init {
services.registerServiceFlow(Requester::class.java, ::Receiver)
}
}
@CordaSerializable
data class OfferMessage(val notary: Party, val dealBeingOffered: IRSState)
@ -52,6 +46,7 @@ object IRSTradeFlow {
}
@InitiatedBy(Requester::class)
class Receiver(private val replyToParty: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() {

11
samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/flows/SimmFlow.kt
View File

@ -11,6 +11,7 @@ import com.opengamma.strata.pricer.swap.DiscountingSwapProductPricer
import net.corda.core.contracts.StateAndRef
import net.corda.core.contracts.StateRef
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.StartableByRPC
import net.corda.core.identity.Party
@ -180,18 +181,10 @@ object SimmFlow {
}
}
/**
* Service plugin for listening for incoming Simm flow communication
*/
class Service(services: PluginServiceHub) {
init {
services.registerServiceFlow(Requester::class.java, ::Receiver)
}
}
/**
* Receives and validates a portfolio and comes to consensus over the portfolio initial margin using SIMM.
*/
@InitiatedBy(Requester::class)
class Receiver(val replyToParty: Party) : FlowLogic<Unit>() {
lateinit var ownParty: Party
lateinit var offer: OfferMessage

3
samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/services/SimmService.kt
View File

@ -15,8 +15,6 @@ import net.corda.core.serialization.SerializationCustomization
import net.corda.vega.analytics.CordaMarketData
import net.corda.vega.analytics.InitialMarginTriple
import net.corda.vega.api.PortfolioApi
import net.corda.vega.flows.IRSTradeFlow
import net.corda.vega.flows.SimmFlow
import java.util.function.Function
/**
@ -28,7 +26,6 @@ object SimmService {
class Plugin : CordaPluginRegistry() {
override val webApis = listOf(Function(::PortfolioApi))
override val staticServeDirs: Map<String, String> = mapOf("simmvaluationdemo" to javaClass.classLoader.getResource("simmvaluationweb").toExternalForm())
override val servicePlugins = listOf(Function(SimmFlow::Service), Function(IRSTradeFlow::Service))
override fun customizeSerialization(custom: SerializationCustomization): Boolean {
custom.apply {
// OpenGamma classes.

11
samples/trader-demo/src/integration-test/kotlin/net/corda/traderdemo/TraderDemoTest.kt
View File

@ -16,6 +16,7 @@ import net.corda.node.services.transactions.SimpleNotaryService
import net.corda.nodeapi.User
import net.corda.testing.BOC
import net.corda.testing.node.NodeBasedTest
import net.corda.traderdemo.flow.BuyerFlow
import net.corda.traderdemo.flow.SellerFlow
import org.assertj.core.api.Assertions.assertThat
import org.junit.Test
@ -36,6 +37,8 @@ class TraderDemoTest : NodeBasedTest() {
startNode(DUMMY_NOTARY.name, advertisedServices = setOf(ServiceInfo(SimpleNotaryService.type)))
).getOrThrow()
nodeA.registerInitiatedFlow(BuyerFlow::class.java)
val (nodeARpc, nodeBRpc) = listOf(nodeA, nodeB).map {
val client = CordaRPCClient(it.configuration.rpcAddress!!)
client.start(demoUser[0].username, demoUser[0].password).proxy
@ -57,12 +60,8 @@ class TraderDemoTest : NodeBasedTest() {
val executor = Executors.newScheduledThreadPool(1)
poll(executor, "A to be notified of the commercial paper", pollInterval = 100.millis) {
val actualPaper = listOf(clientA.commercialPaperCount, clientB.commercialPaperCount)
if (actualPaper == expectedPaper) {
Unit
} else {
null
}
}.get()
if (actualPaper == expectedPaper) Unit else null
}.getOrThrow()
executor.shutdown()
assertThat(clientA.dollarCashBalance).isEqualTo(95.DOLLARS)
assertThat(clientB.dollarCashBalance).isEqualTo(5.DOLLARS)

33
samples/trader-demo/src/main/kotlin/net/corda/traderdemo/flow/BuyerFlow.kt
View File

@ -4,36 +4,24 @@ import co.paralleluniverse.fibers.Suspendable
import net.corda.contracts.CommercialPaper
import net.corda.core.contracts.Amount
import net.corda.core.contracts.TransactionGraphSearch
import net.corda.core.identity.Party
import net.corda.core.div
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.identity.Party
import net.corda.core.node.NodeInfo
import net.corda.core.node.PluginServiceHub
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.Emoji
import net.corda.core.utilities.ProgressTracker
import net.corda.core.utilities.unwrap
import net.corda.flows.TwoPartyTradeFlow
import java.nio.file.Paths
import java.util.*
class BuyerFlow(val otherParty: Party,
private val attachmentsDirectory: String,
override val progressTracker: ProgressTracker = ProgressTracker(STARTING_BUY)) : FlowLogic<Unit>() {
@InitiatedBy(SellerFlow::class)
class BuyerFlow(val otherParty: Party) : FlowLogic<Unit>() {
object STARTING_BUY : ProgressTracker.Step("Seller connected, purchasing commercial paper asset")
class Service(services: PluginServiceHub) : SingletonSerializeAsToken() {
init {
// Buyer will fetch the attachment from the seller automatically when it resolves the transaction.
// For demo purposes just extract attachment jars when saved to disk, so the user can explore them.
val attachmentsPath = (services.storageService.attachments).let {
it.automaticallyExtractAttachments = true
it.storePath
}
services.registerServiceFlow(SellerFlow::class.java) { BuyerFlow(it, attachmentsPath.toString()) }
}
}
override val progressTracker: ProgressTracker = ProgressTracker(STARTING_BUY)
@Suspendable
override fun call() {
@ -72,8 +60,15 @@ class BuyerFlow(val otherParty: Party,
followInputsOfType = CommercialPaper.State::class.java)
val cpIssuance = search.call().single()
// Buyer will fetch the attachment from the seller automatically when it resolves the transaction.
// For demo purposes just extract attachment jars when saved to disk, so the user can explore them.
val attachmentsPath = (serviceHub.storageService.attachments).let {
it.automaticallyExtractAttachments = true
it.storePath
}
cpIssuance.attachments.first().let {
val p = Paths.get(attachmentsDirectory, "$it.jar")
val p = attachmentsPath / "$it.jar"
println("""
The issuance of the commercial paper came with an attachment. You can find it expanded in this directory:

10
samples/trader-demo/src/main/kotlin/net/corda/traderdemo/plugin/TraderDemoPlugin.kt
View File

@ -1,10 +0,0 @@
package net.corda.traderdemo.plugin
import net.corda.core.identity.Party
import net.corda.core.node.CordaPluginRegistry
import net.corda.traderdemo.flow.BuyerFlow
import java.util.function.Function
class TraderDemoPlugin : CordaPluginRegistry() {
override val servicePlugins = listOf(Function(BuyerFlow::Service))
}

2
samples/trader-demo/src/main/resources/META-INF/services/net.corda.core.node.CordaPluginRegistry
View File

@ -1,2 +0,0 @@
# Register a ServiceLoader service extending from net.corda.core.node.CordaPluginRegistry
net.corda.traderdemo.plugin.TraderDemoPlugin

25
test-utils/src/main/kotlin/net/corda/testing/CoreTestUtils.kt
View File

@ -4,24 +4,21 @@
package net.corda.testing
import com.google.common.net.HostAndPort
import com.google.common.util.concurrent.ListenableFuture
import net.corda.core.contracts.StateRef
import net.corda.core.crypto.*
import net.corda.core.flows.FlowLogic
import net.corda.core.crypto.SecureHash
import net.corda.core.crypto.X509Utilities
import net.corda.core.crypto.commonName
import net.corda.core.crypto.generateKeyPair
import net.corda.core.identity.Party
import net.corda.core.node.ServiceHub
import net.corda.core.node.VersionInfo
import net.corda.core.node.services.IdentityService
import net.corda.core.serialization.OpaqueBytes
import net.corda.core.toFuture
import net.corda.core.transactions.TransactionBuilder
import net.corda.core.utilities.*
import net.corda.node.internal.AbstractNode
import net.corda.node.internal.NetworkMapInfo
import net.corda.node.services.config.*
import net.corda.node.services.identity.InMemoryIdentityService
import net.corda.node.services.statemachine.FlowStateMachineImpl
import net.corda.node.services.statemachine.StateMachineManager
import net.corda.nodeapi.User
import net.corda.nodeapi.config.SSLConfiguration
import net.corda.testing.node.MockServices
@ -36,7 +33,6 @@ import java.security.KeyPair
import java.security.PublicKey
import java.util.*
import java.util.concurrent.atomic.AtomicInteger
import kotlin.reflect.KClass
/**
* JAVA INTEROP
@ -138,19 +134,6 @@ fun getFreeLocalPorts(hostName: String, numberToAlloc: Int): List<HostAndPort> {
dsl: TransactionDSL<TransactionDSLInterpreter>.() -> EnforceVerifyOrFail
) = ledger { this.transaction(transactionLabel, transactionBuilder, dsl) }
/**
* The given flow factory will be used to initiate just one instance of a flow of type [P] when a counterparty
* flow requests for it using [clientFlowClass].
* @return Returns a [ListenableFuture] holding the single [FlowStateMachineImpl] created by the request.
*/
inline fun <reified P : FlowLogic<*>> AbstractNode.initiateSingleShotFlow(
clientFlowClass: KClass<out FlowLogic<*>>,
noinline serviceFlowFactory: (Party) -> P): ListenableFuture<P> {
val future = smm.changes.filter { it is StateMachineManager.Change.Add && it.logic is P }.map { it.logic as P }.toFuture()
services.registerServiceFlow(clientFlowClass.java, serviceFlowFactory)
return future
}
// TODO Replace this with testConfiguration
data class TestNodeConfiguration(
override val baseDirectory: Path,

14
test-utils/src/main/kotlin/net/corda/testing/node/MockNode.kt
View File

@ -1,13 +1,11 @@
package net.corda.testing.node
import com.google.common.annotations.VisibleForTesting
import com.google.common.jimfs.Configuration.unix
import com.google.common.jimfs.Jimfs
import com.google.common.util.concurrent.Futures
import com.google.common.util.concurrent.ListenableFuture
import net.corda.core.*
import net.corda.core.crypto.entropyToKeyPair
import net.corda.core.flows.FlowLogic
import net.corda.core.identity.Party
import net.corda.core.messaging.RPCOps
import net.corda.core.messaging.SingleMessageRecipient
@ -18,14 +16,12 @@ import net.corda.core.node.services.*
import net.corda.core.utilities.DUMMY_NOTARY_KEY
import net.corda.core.utilities.loggerFor
import net.corda.node.internal.AbstractNode
import net.corda.node.internal.ServiceFlowInfo
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.identity.InMemoryIdentityService
import net.corda.node.services.keys.E2ETestKeyManagementService
import net.corda.node.services.messaging.MessagingService
import net.corda.node.services.network.InMemoryNetworkMapService
import net.corda.node.services.network.NetworkMapService
import net.corda.node.services.statemachine.flowVersion
import net.corda.node.services.transactions.InMemoryTransactionVerifierService
import net.corda.node.services.transactions.InMemoryUniquenessProvider
import net.corda.node.services.transactions.SimpleNotaryService
@ -45,7 +41,6 @@ import java.security.KeyPair
import java.util.*
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import kotlin.reflect.KClass
/**
* A mock node brings up a suite of in-memory services in a fast manner suitable for unit testing.
@ -232,13 +227,6 @@ class MockNetwork(private val networkSendManuallyPumped: Boolean = false,
// It is used from the network visualiser tool.
@Suppress("unused") val place: PhysicalLocation get() = findMyLocation()!!
@VisibleForTesting
fun registerServiceFlow(clientFlowClass: KClass<out FlowLogic<*>>,
flowVersion: Int = clientFlowClass.java.flowVersion,
serviceFlowFactory: (Party) -> FlowLogic<*>) {
serviceFlowFactories[clientFlowClass.java] = ServiceFlowInfo.CorDapp(flowVersion, serviceFlowFactory)
}
fun pumpReceive(block: Boolean = false): InMemoryMessagingNetwork.MessageTransfer? {
return (net as InMemoryMessagingNetwork.InMemoryMessaging).pumpReceive(block)
}
@ -262,8 +250,6 @@ class MockNetwork(private val networkSendManuallyPumped: Boolean = false,
* Returns a node, optionally created by the passed factory method.
* @param overrideServices a set of service entries to use in place of the node's default service entries,
* for example where a node's service is part of a cluster.
* @param entropyRoot the initial entropy value to use when generating keys. Defaults to an (insecure) random value,
* but can be overriden to cause nodes to have stable or colliding identity/service keys.
*/
fun createNode(networkMapAddress: SingleMessageRecipient? = null, forcedID: Int = -1, nodeFactory: Factory = defaultFactory,
start: Boolean = true, legalName: X500Name? = null, overrideServices: Map<ServiceInfo, KeyPair>? = null,

10
test-utils/src/main/kotlin/net/corda/testing/node/MockServices.kt
View File

@ -8,6 +8,7 @@ import net.corda.core.messaging.SingleMessageRecipient
import net.corda.core.node.NodeInfo
import net.corda.core.node.ServiceHub
import net.corda.core.node.services.*
import net.corda.core.serialization.SerializeAsToken
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.DUMMY_NOTARY
@ -28,6 +29,7 @@ import java.io.ByteArrayInputStream
import java.io.ByteArrayOutputStream
import java.io.File
import java.io.InputStream
import java.nio.file.Path
import java.nio.file.Paths
import java.security.KeyPair
import java.security.PrivateKey
@ -75,6 +77,8 @@ open class MockServices(vararg val keys: KeyPair) : ServiceHub {
HibernateObserver(vaultService.rawUpdates, NodeSchemaService())
return vaultService
}
override fun <T : SerializeAsToken> cordaService(type: Class<T>): T = throw IllegalArgumentException("${type.name} not found")
}
class MockKeyManagementService(val identityService: IdentityService,
@ -91,7 +95,9 @@ class MockKeyManagementService(val identityService: IdentityService,
return k.public
}
override fun freshKeyAndCert(identity: Party, revocationEnabled: Boolean): Pair<X509CertificateHolder, CertPath> = freshKeyAndCert(this, identityService, identity, revocationEnabled)
override fun freshKeyAndCert(identity: Party, revocationEnabled: Boolean): Pair<X509CertificateHolder, CertPath> {
return freshKeyAndCert(this, identityService, identity, revocationEnabled)
}
private fun getSigningKeyPair(publicKey: PublicKey): KeyPair {
val pk = publicKey.keys.first { keyStore.containsKey(it) }
@ -108,7 +114,7 @@ class MockKeyManagementService(val identityService: IdentityService,
class MockAttachmentStorage : AttachmentStorage {
val files = HashMap<SecureHash, ByteArray>()
override var automaticallyExtractAttachments = false
override var storePath = Paths.get("")
override var storePath: Path = Paths.get("")
override fun openAttachment(id: SecureHash): Attachment? {
val f = files[id] ?: return null

10
tools/explorer/src/main/kotlin/net/corda/explorer/plugin/ExplorerPlugin.kt
View File

@ -1,10 +0,0 @@
package net.corda.explorer.plugin
import net.corda.core.identity.Party
import net.corda.core.node.CordaPluginRegistry
import net.corda.flows.IssuerFlow
import java.util.function.Function
class ExplorerPlugin : CordaPluginRegistry() {
override val servicePlugins = listOf(Function(IssuerFlow.Issuer::Service))
}

2
tools/explorer/src/main/resources/META-INF/services/net.corda.core.node.CordaPluginRegistry
View File

@ -1,2 +0,0 @@
# Register a ServiceLoader service extending from net.corda.node.CordaPluginRegistry
net.corda.explorer.plugin.ExplorerPlugin