mirror of
https://github.com/corda/corda.git
synced 2025-06-17 06:38:21 +00:00
Introduced FlowLogic.getFlowContext which provides the flow version and app name of the other side.
This commit is contained in:
Shams Asari
@ -6,35 +6,40 @@ import net.corda.core.flows.InitiatingFlow
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.concurrent.transpose
|
||||
import net.corda.core.utilities.getOrThrow
|
||||
import net.corda.core.utilities.unwrap
|
||||
import net.corda.testing.ALICE
|
||||
import net.corda.testing.BOB
|
||||
import net.corda.core.utilities.unwrap
|
||||
import net.corda.testing.node.NodeBasedTest
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.Test
|
||||
|
||||
class FlowVersioningTest : NodeBasedTest() {
|
||||
@Test
|
||||
fun `core flows receive platform version of initiator`() {
|
||||
fun `getFlowContext returns the platform version for core flows`() {
|
||||
val (alice, bob) = listOf(
|
||||
startNode(ALICE.name, platformVersion = 2),
|
||||
startNode(BOB.name, platformVersion = 3)).transpose().getOrThrow()
|
||||
bob.installCoreFlow(ClientFlow::class, ::SendBackPlatformVersionFlow)
|
||||
val resultFuture = alice.services.startFlow(ClientFlow(bob.info.legalIdentity)).resultFuture
|
||||
assertThat(resultFuture.getOrThrow()).isEqualTo(2)
|
||||
bob.installCoreFlow(PretendInitiatingCoreFlow::class, ::PretendInitiatedCoreFlow)
|
||||
val (alicePlatformVersionAccordingToBob, bobPlatformVersionAccordingToAlice) = alice.services.startFlow(
|
||||
PretendInitiatingCoreFlow(bob.info.legalIdentity)).resultFuture.getOrThrow()
|
||||
assertThat(alicePlatformVersionAccordingToBob).isEqualTo(2)
|
||||
assertThat(bobPlatformVersionAccordingToAlice).isEqualTo(3)
|
||||
}
|
||||
|
||||
@InitiatingFlow
|
||||
private class ClientFlow(val otherParty: Party) : FlowLogic<Any>() {
|
||||
private class PretendInitiatingCoreFlow(val initiatedParty: Party) : FlowLogic<Pair<Int, Int>>() {
|
||||
@Suspendable
|
||||
override fun call(): Any {
|
||||
return sendAndReceive<Any>(otherParty, "This is ignored. We only send to kick off the flow on the other side").unwrap { it }
|
||||
override fun call(): Pair<Int, Int> {
|
||||
return Pair(
|
||||
receive<Int>(initiatedParty).unwrap { it },
|
||||
getFlowContext(initiatedParty).flowVersion
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private class SendBackPlatformVersionFlow(val otherParty: Party, val otherPartysPlatformVersion: Int) : FlowLogic<Unit>() {
|
||||
private class PretendInitiatedCoreFlow(val initiatingParty: Party) : FlowLogic<Unit>() {
|
||||
@Suspendable
|
||||
override fun call() = send(otherParty, otherPartysPlatformVersion)
|
||||
override fun call() = send(initiatingParty, getFlowContext(initiatingParty).flowVersion)
|
||||
}
|
||||
|
||||
}
|
||||
@ -30,7 +30,10 @@ import net.corda.flows.CashExitFlow
|
||||
import net.corda.flows.CashIssueFlow
|
||||
import net.corda.flows.CashPaymentFlow
|
||||
import net.corda.flows.IssuerFlow
|
||||
import net.corda.node.services.*
|
||||
import net.corda.node.services.ContractUpgradeHandler
|
||||
import net.corda.node.services.NotaryChangeHandler
|
||||
import net.corda.node.services.NotifyTransactionHandler
|
||||
import net.corda.node.services.TransactionKeyHandler
|
||||
import net.corda.node.services.api.*
|
||||
import net.corda.node.services.config.NodeConfiguration
|
||||
import net.corda.node.services.config.configureWithDevSSLCertificate
|
||||
@ -62,8 +65,6 @@ import net.corda.node.services.vault.NodeVaultService
|
||||
import net.corda.node.services.vault.VaultSoftLockManager
|
||||
import net.corda.node.utilities.*
|
||||
import net.corda.node.utilities.AddOrRemove.ADD
|
||||
import net.corda.node.utilities.AffinityExecutor
|
||||
import net.corda.node.utilities.configureDatabase
|
||||
import org.apache.activemq.artemis.utils.ReusableLatch
|
||||
import org.bouncycastle.asn1.x500.X500Name
|
||||
import org.slf4j.Logger
|
||||
@ -282,7 +283,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
private fun handleCustomNotaryService(service: NotaryService) {
|
||||
runOnStop += service::stop
|
||||
service.start()
|
||||
installCoreFlow(NotaryFlow.Client::class, { party: Party, version: Int -> service.createServiceFlow(party, version) })
|
||||
installCoreFlow(NotaryFlow.Client::class, service::createServiceFlow)
|
||||
}
|
||||
|
||||
private inline fun <reified A : Annotation> Class<*>.requireAnnotation(): A {
|
||||
@ -344,9 +345,15 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
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"
|
||||
"${InitiatedBy::class.java.name} must point to ${classWithAnnotation.name} and not ${initiatingFlow.name}"
|
||||
}
|
||||
val flowFactory = InitiatedFlowFactory.CorDapp(version, { ctor.newInstance(it) })
|
||||
val jarFile = Paths.get(initiatedFlow.protectionDomain.codeSource.location.toURI())
|
||||
val appName = if (jarFile.isRegularFile() && jarFile.toString().endsWith(".jar")) {
|
||||
jarFile.fileName.toString().removeSuffix(".jar")
|
||||
} else {
|
||||
"<unknown>"
|
||||
}
|
||||
val flowFactory = InitiatedFlowFactory.CorDapp(version, appName, { ctor.newInstance(it) })
|
||||
val observable = internalRegisterFlowFactory(initiatingFlow, flowFactory, initiatedFlow, track)
|
||||
log.info("Registered ${initiatingFlow.name} to initiate ${initiatedFlow.name} (version $version)")
|
||||
return observable
|
||||
@ -390,7 +397,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
* @suppress
|
||||
*/
|
||||
@VisibleForTesting
|
||||
fun installCoreFlow(clientFlowClass: KClass<out FlowLogic<*>>, flowFactory: (Party, Int) -> FlowLogic<*>) {
|
||||
fun installCoreFlow(clientFlowClass: KClass<out FlowLogic<*>>, flowFactory: (Party) -> 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"
|
||||
}
|
||||
@ -399,10 +406,10 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
}
|
||||
|
||||
private fun installCoreFlows() {
|
||||
installCoreFlow(BroadcastTransactionFlow::class) { otherParty, _ -> NotifyTransactionHandler(otherParty) }
|
||||
installCoreFlow(NotaryChangeFlow::class) { otherParty, _ -> NotaryChangeHandler(otherParty) }
|
||||
installCoreFlow(ContractUpgradeFlow::class) { otherParty, _ -> ContractUpgradeHandler(otherParty) }
|
||||
installCoreFlow(TransactionKeyFlow::class) { otherParty, _ -> TransactionKeyHandler(otherParty) }
|
||||
installCoreFlow(BroadcastTransactionFlow::class, ::NotifyTransactionHandler)
|
||||
installCoreFlow(NotaryChangeFlow::class, ::NotaryChangeHandler)
|
||||
installCoreFlow(ContractUpgradeFlow::class, ::ContractUpgradeHandler)
|
||||
installCoreFlow(TransactionKeyFlow::class, ::TransactionKeyHandler)
|
||||
}
|
||||
|
||||
/**
|
||||
@ -567,7 +574,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
runOnStop += this::stop
|
||||
start()
|
||||
}
|
||||
installCoreFlow(NotaryFlow.Client::class, { party: Party, version: Int -> service.createServiceFlow(party, version) })
|
||||
installCoreFlow(NotaryFlow.Client::class, service::createServiceFlow)
|
||||
} else {
|
||||
log.info("Notary type ${notaryServiceType.id} does not match any built-in notary types. " +
|
||||
"It is expected to be loaded via a CorDapp")
|
||||
|
||||
@ -2,26 +2,14 @@ package net.corda.node.internal
|
||||
|
||||
import net.corda.core.flows.FlowLogic
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.node.services.statemachine.SessionInit
|
||||
import net.corda.node.services.statemachine.SessionRejectException
|
||||
|
||||
interface InitiatedFlowFactory<out F : FlowLogic<*>> {
|
||||
fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): F
|
||||
sealed class InitiatedFlowFactory<out F : FlowLogic<*>> {
|
||||
protected abstract val factory: (Party) -> F
|
||||
fun createFlow(otherParty: Party): F = factory(otherParty)
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
data class Core<out F : FlowLogic<*>>(override val factory: (Party) -> F) : InitiatedFlowFactory<F>()
|
||||
data class CorDapp<out F : FlowLogic<*>>(val flowVersion: Int,
|
||||
val appName: String,
|
||||
override val factory: (Party) -> F) : InitiatedFlowFactory<F>()
|
||||
}
|
||||
|
||||
|
||||
@ -1,7 +1,8 @@
|
||||
package net.corda.node.services.statemachine
|
||||
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.flows.FlowContext
|
||||
import net.corda.core.flows.FlowLogic
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.node.services.statemachine.FlowSessionState.Initiated
|
||||
import net.corda.node.services.statemachine.FlowSessionState.Initiating
|
||||
import java.util.concurrent.ConcurrentLinkedQueue
|
||||
@ -41,7 +42,7 @@ sealed class FlowSessionState {
|
||||
override val sendToParty: Party get() = otherParty
|
||||
}
|
||||
|
||||
data class Initiated(val peerParty: Party, val peerSessionId: Long) : FlowSessionState() {
|
||||
data class Initiated(val peerParty: Party, val peerSessionId: Long, val context: FlowContext) : FlowSessionState() {
|
||||
override val sendToParty: Party get() = peerParty
|
||||
}
|
||||
}
|
||||
|
||||
@ -10,9 +10,9 @@ import net.corda.core.crypto.random63BitValue
|
||||
import net.corda.core.flows.*
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.FlowStateMachine
|
||||
import net.corda.core.internal.abbreviate
|
||||
import net.corda.core.internal.concurrent.OpenFuture
|
||||
import net.corda.core.internal.concurrent.openFuture
|
||||
import net.corda.core.internal.abbreviate
|
||||
import net.corda.core.internal.staticField
|
||||
import net.corda.core.transactions.SignedTransaction
|
||||
import net.corda.core.utilities.*
|
||||
@ -25,7 +25,6 @@ import net.corda.node.utilities.DatabaseTransaction
|
||||
import net.corda.node.utilities.DatabaseTransactionManager
|
||||
import org.slf4j.Logger
|
||||
import org.slf4j.LoggerFactory
|
||||
import java.sql.Connection
|
||||
import java.sql.SQLException
|
||||
import java.util.*
|
||||
import java.util.concurrent.TimeUnit
|
||||
@ -154,6 +153,12 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
}
|
||||
}
|
||||
|
||||
@Suspendable
|
||||
override fun getFlowContext(otherParty: Party, sessionFlow: FlowLogic<*>): FlowContext {
|
||||
val state = getConfirmedSession(otherParty, sessionFlow).state as FlowSessionState.Initiated
|
||||
return state.context
|
||||
}
|
||||
|
||||
@Suspendable
|
||||
override fun <T : Any> sendAndReceive(receiveType: Class<T>,
|
||||
otherParty: Party,
|
||||
@ -161,7 +166,7 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
sessionFlow: FlowLogic<*>,
|
||||
retrySend: Boolean): UntrustworthyData<T> {
|
||||
logger.debug { "sendAndReceive(${receiveType.name}, $otherParty, ${payload.toString().abbreviate(300)}) ..." }
|
||||
val session = getConfirmedSession(otherParty, sessionFlow)
|
||||
val session = getConfirmedSessionIfPresent(otherParty, sessionFlow)
|
||||
val sessionData = if (session == null) {
|
||||
val newSession = startNewSession(otherParty, sessionFlow, payload, waitForConfirmation = true, retryable = retrySend)
|
||||
// Only do a receive here as the session init has carried the payload
|
||||
@ -179,8 +184,7 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
otherParty: Party,
|
||||
sessionFlow: FlowLogic<*>): UntrustworthyData<T> {
|
||||
logger.debug { "receive(${receiveType.name}, $otherParty) ..." }
|
||||
val session = getConfirmedSession(otherParty, sessionFlow) ?:
|
||||
startNewSession(otherParty, sessionFlow, null, waitForConfirmation = true)
|
||||
val session = getConfirmedSession(otherParty, sessionFlow)
|
||||
val sessionData = receiveInternal<SessionData>(session, receiveType)
|
||||
logger.debug { "Received ${sessionData.message.payload.toString().abbreviate(300)}" }
|
||||
return sessionData.checkPayloadIs(receiveType)
|
||||
@ -189,7 +193,7 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
@Suspendable
|
||||
override fun send(otherParty: Party, payload: Any, sessionFlow: FlowLogic<*>) {
|
||||
logger.debug { "send($otherParty, ${payload.toString().abbreviate(300)})" }
|
||||
val session = getConfirmedSession(otherParty, sessionFlow)
|
||||
val session = getConfirmedSessionIfPresent(otherParty, sessionFlow)
|
||||
if (session == null) {
|
||||
// Don't send the payload again if it was already piggy-backed on a session init
|
||||
startNewSession(otherParty, sessionFlow, payload, waitForConfirmation = false)
|
||||
@ -257,7 +261,10 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
private fun FlowSession.waitForConfirmation() {
|
||||
val (peerParty, sessionInitResponse) = receiveInternal<SessionInitResponse>(this, null)
|
||||
if (sessionInitResponse is SessionConfirm) {
|
||||
state = FlowSessionState.Initiated(peerParty, sessionInitResponse.initiatedSessionId)
|
||||
state = FlowSessionState.Initiated(
|
||||
peerParty,
|
||||
sessionInitResponse.initiatedSessionId,
|
||||
FlowContext(sessionInitResponse.flowVersion, sessionInitResponse.appName))
|
||||
} else {
|
||||
sessionInitResponse as SessionReject
|
||||
throw UnexpectedFlowEndException("Party ${state.sendToParty} rejected session request: ${sessionInitResponse.errorMessage}")
|
||||
@ -274,9 +281,7 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
}
|
||||
|
||||
@Suspendable
|
||||
private fun sendInternal(session: FlowSession, message: SessionMessage) {
|
||||
suspend(SendOnly(session, message))
|
||||
}
|
||||
private fun sendInternal(session: FlowSession, message: SessionMessage) = suspend(SendOnly(session, message))
|
||||
|
||||
private inline fun <reified M : ExistingSessionMessage> receiveInternal(
|
||||
session: FlowSession,
|
||||
@ -292,15 +297,21 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
}
|
||||
|
||||
@Suspendable
|
||||
private fun getConfirmedSession(otherParty: Party, sessionFlow: FlowLogic<*>): FlowSession? {
|
||||
private fun getConfirmedSessionIfPresent(otherParty: Party, sessionFlow: FlowLogic<*>): FlowSession? {
|
||||
return openSessions[Pair(sessionFlow, otherParty)]?.apply {
|
||||
if (state is FlowSessionState.Initiating) {
|
||||
// Session still initiating, try to retrieve the init response.
|
||||
// Session still initiating, wait for the confirmation
|
||||
waitForConfirmation()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Suspendable
|
||||
private fun getConfirmedSession(otherParty: Party, sessionFlow: FlowLogic<*>): FlowSession {
|
||||
return getConfirmedSessionIfPresent(otherParty, sessionFlow) ?:
|
||||
startNewSession(otherParty, sessionFlow, null, waitForConfirmation = true)
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new session. The provided [otherParty] can be an identity of any advertised service on the network,
|
||||
* and might be advertised by more than one node. Therefore we first choose a single node that advertises it
|
||||
@ -317,7 +328,7 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
|
||||
val session = FlowSession(sessionFlow, random63BitValue(), null, FlowSessionState.Initiating(otherParty), retryable)
|
||||
openSessions[Pair(sessionFlow, otherParty)] = session
|
||||
val (version, initiatingFlowClass) = sessionFlow.javaClass.flowVersionAndInitiatingClass
|
||||
val sessionInit = SessionInit(session.ourSessionId, initiatingFlowClass.name, version, firstPayload)
|
||||
val sessionInit = SessionInit(session.ourSessionId, initiatingFlowClass.name, version, "not defined", firstPayload)
|
||||
sendInternal(session, sessionInit)
|
||||
if (waitForConfirmation) {
|
||||
session.waitForConfirmation()
|
||||
@ -456,6 +467,7 @@ val Class<out FlowLogic<*>>.flowVersionAndInitiatingClass: Pair<Int, Class<out F
|
||||
}
|
||||
current = current.superclass
|
||||
?: return found
|
||||
?: throw IllegalArgumentException("$name, as a flow that initiates other flows, must be annotated with ${InitiatingFlow::class.java.name}. See https://docs.corda.net/api-flows.html#flowlogic-annotations.")
|
||||
?: throw IllegalArgumentException("$name, as a flow that initiates other flows, must be annotated with " +
|
||||
"${InitiatingFlow::class.java.name}. See https://docs.corda.net/api-flows.html#flowlogic-annotations.")
|
||||
}
|
||||
}
|
||||
|
||||
@ -7,36 +7,37 @@ import net.corda.core.internal.castIfPossible
|
||||
import net.corda.core.serialization.CordaSerializable
|
||||
import net.corda.core.utilities.UntrustworthyData
|
||||
|
||||
/**
|
||||
* These internal messages define the flow session protocol.
|
||||
*/
|
||||
|
||||
@CordaSerializable
|
||||
interface SessionMessage
|
||||
|
||||
data class SessionInit(val initiatorSessionId: Long,
|
||||
val initiatingFlowClass: String,
|
||||
val flowVerison: Int,
|
||||
val firstPayload: Any?) : SessionMessage
|
||||
|
||||
interface ExistingSessionMessage : SessionMessage {
|
||||
val recipientSessionId: Long
|
||||
}
|
||||
|
||||
data class SessionData(override val recipientSessionId: Long, val payload: Any) : ExistingSessionMessage {
|
||||
override fun toString(): String = "${javaClass.simpleName}(recipientSessionId=$recipientSessionId, payload=$payload)"
|
||||
}
|
||||
|
||||
interface SessionInitResponse : ExistingSessionMessage {
|
||||
val initiatorSessionId: Long
|
||||
override val recipientSessionId: Long get() = initiatorSessionId
|
||||
}
|
||||
|
||||
data class SessionConfirm(override val initiatorSessionId: Long, val initiatedSessionId: Long) : SessionInitResponse
|
||||
interface SessionEnd : ExistingSessionMessage
|
||||
|
||||
data class SessionInit(val initiatorSessionId: Long,
|
||||
val initiatingFlowClass: String,
|
||||
val flowVersion: Int,
|
||||
val appName: String,
|
||||
val firstPayload: Any?) : SessionMessage
|
||||
|
||||
data class SessionConfirm(override val initiatorSessionId: Long,
|
||||
val initiatedSessionId: Long,
|
||||
val flowVersion: Int,
|
||||
val appName: String) : SessionInitResponse
|
||||
|
||||
data class SessionReject(override val initiatorSessionId: Long, val errorMessage: String) : SessionInitResponse
|
||||
|
||||
interface SessionEnd : ExistingSessionMessage
|
||||
data class SessionData(override val recipientSessionId: Long, val payload: Any) : ExistingSessionMessage
|
||||
|
||||
data class NormalSessionEnd(override val recipientSessionId: Long) : SessionEnd
|
||||
|
||||
data class ErrorSessionEnd(override val recipientSessionId: Long, val errorResponse: FlowException?) : SessionEnd
|
||||
|
||||
data class ReceivedSessionMessage<out M : ExistingSessionMessage>(val sender: Party, val message: M)
|
||||
|
||||
@ -10,10 +10,7 @@ import com.google.common.util.concurrent.MoreExecutors
|
||||
import net.corda.core.concurrent.CordaFuture
|
||||
import net.corda.core.crypto.SecureHash
|
||||
import net.corda.core.crypto.random63BitValue
|
||||
import net.corda.core.flows.FlowException
|
||||
import net.corda.core.flows.FlowInitiator
|
||||
import net.corda.core.flows.FlowLogic
|
||||
import net.corda.core.flows.StateMachineRunId
|
||||
import net.corda.core.flows.*
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.ThreadBox
|
||||
import net.corda.core.internal.bufferUntilSubscribed
|
||||
@ -28,6 +25,7 @@ import net.corda.core.utilities.Try
|
||||
import net.corda.core.utilities.debug
|
||||
import net.corda.core.utilities.loggerFor
|
||||
import net.corda.core.utilities.trace
|
||||
import net.corda.node.internal.InitiatedFlowFactory
|
||||
import net.corda.node.services.api.Checkpoint
|
||||
import net.corda.node.services.api.CheckpointStorage
|
||||
import net.corda.node.services.api.ServiceHubInternal
|
||||
@ -205,7 +203,7 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
|
||||
* @param allowedUnsuspendedFiberCount Optional parameter is used in some tests.
|
||||
*/
|
||||
fun stop(allowedUnsuspendedFiberCount: Int = 0) {
|
||||
check(allowedUnsuspendedFiberCount >= 0)
|
||||
require(allowedUnsuspendedFiberCount >= 0)
|
||||
mutex.locked {
|
||||
if (stopping) throw IllegalStateException("Already stopping!")
|
||||
stopping = true
|
||||
@ -340,23 +338,30 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
|
||||
|
||||
private fun onSessionInit(sessionInit: SessionInit, receivedMessage: ReceivedMessage, sender: Party) {
|
||||
logger.trace { "Received $sessionInit from $sender" }
|
||||
val otherPartySessionId = sessionInit.initiatorSessionId
|
||||
val senderSessionId = sessionInit.initiatorSessionId
|
||||
|
||||
fun sendSessionReject(message: String) = sendSessionMessage(sender, SessionReject(otherPartySessionId, message))
|
||||
fun sendSessionReject(message: String) = sendSessionMessage(sender, SessionReject(senderSessionId, message))
|
||||
|
||||
val session = try {
|
||||
val initiatedFlowFactory = serviceHub.getFlowFactory(sessionInit.loadInitiatingFlowClass())
|
||||
?: throw SessionRejectException("${sessionInit.initiatingFlowClass} is not registered")
|
||||
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))
|
||||
val (session, initiatedFlowFactory) = try {
|
||||
val initiatedFlowFactory = getInitiatedFlowFactory(sessionInit)
|
||||
val flow = initiatedFlowFactory.createFlow(sender)
|
||||
val senderFlowVersion = when (initiatedFlowFactory) {
|
||||
is InitiatedFlowFactory.Core -> receivedMessage.platformVersion // The flow version for the core flows is the platform version
|
||||
is InitiatedFlowFactory.CorDapp -> sessionInit.flowVersion
|
||||
}
|
||||
val session = FlowSession(
|
||||
flow,
|
||||
random63BitValue(),
|
||||
sender,
|
||||
FlowSessionState.Initiated(sender, senderSessionId, FlowContext(senderFlowVersion, sessionInit.appName)))
|
||||
if (sessionInit.firstPayload != null) {
|
||||
session.receivedMessages += ReceivedSessionMessage(sender, SessionData(session.ourSessionId, sessionInit.firstPayload))
|
||||
}
|
||||
openSessions[session.ourSessionId] = session
|
||||
val fiber = createFiber(flow, FlowInitiator.Peer(sender))
|
||||
fiber.openSessions[Pair(flow, sender)] = session
|
||||
updateCheckpoint(fiber)
|
||||
session
|
||||
session to initiatedFlowFactory
|
||||
} catch (e: SessionRejectException) {
|
||||
logger.warn("${e.logMessage}: $sessionInit")
|
||||
sendSessionReject(e.rejectMessage)
|
||||
@ -367,20 +372,28 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
|
||||
return
|
||||
}
|
||||
|
||||
sendSessionMessage(sender, SessionConfirm(otherPartySessionId, session.ourSessionId), session.fiber)
|
||||
val (ourFlowVersion, appName) = when (initiatedFlowFactory) {
|
||||
// The flow version for the core flows is the platform version
|
||||
is InitiatedFlowFactory.Core -> serviceHub.myInfo.platformVersion to "corda"
|
||||
is InitiatedFlowFactory.CorDapp -> initiatedFlowFactory.flowVersion to initiatedFlowFactory.appName
|
||||
}
|
||||
|
||||
sendSessionMessage(sender, SessionConfirm(senderSessionId, session.ourSessionId, ourFlowVersion, appName), session.fiber)
|
||||
session.fiber.logger.debug { "Initiated by $sender using ${sessionInit.initiatingFlowClass}" }
|
||||
session.fiber.logger.trace { "Initiated from $sessionInit on $session" }
|
||||
resumeFiber(session.fiber)
|
||||
}
|
||||
|
||||
private fun SessionInit.loadInitiatingFlowClass(): Class<out FlowLogic<*>> {
|
||||
return try {
|
||||
Class.forName(initiatingFlowClass).asSubclass(FlowLogic::class.java)
|
||||
private fun getInitiatedFlowFactory(sessionInit: SessionInit): InitiatedFlowFactory<*> {
|
||||
val initiatingFlowClass = try {
|
||||
Class.forName(sessionInit.initiatingFlowClass).asSubclass(FlowLogic::class.java)
|
||||
} catch (e: ClassNotFoundException) {
|
||||
throw SessionRejectException("Don't know $initiatingFlowClass")
|
||||
throw SessionRejectException("Don't know ${sessionInit.initiatingFlowClass}")
|
||||
} catch (e: ClassCastException) {
|
||||
throw SessionRejectException("$initiatingFlowClass is not a flow")
|
||||
throw SessionRejectException("${sessionInit.initiatingFlowClass} is not a flow")
|
||||
}
|
||||
return serviceHub.getFlowFactory(initiatingFlowClass) ?:
|
||||
throw SessionRejectException("$initiatingFlowClass is not registered")
|
||||
}
|
||||
|
||||
private fun serializeFiber(fiber: FlowStateMachineImpl<*>): SerializedBytes<FlowStateMachineImpl<*>> {
|
||||
@ -389,7 +402,9 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
|
||||
|
||||
private fun deserializeFiber(checkpoint: Checkpoint, logger: Logger): FlowStateMachineImpl<*>? {
|
||||
return try {
|
||||
checkpoint.serializedFiber.deserialize(context = CHECKPOINT_CONTEXT.withTokenContext(serializationContext)).apply { fromCheckpoint = true }
|
||||
checkpoint.serializedFiber.deserialize(context = CHECKPOINT_CONTEXT.withTokenContext(serializationContext)).apply {
|
||||
fromCheckpoint = true
|
||||
}
|
||||
} catch (t: Throwable) {
|
||||
logger.error("Encountered unrestorable checkpoint!", t)
|
||||
null
|
||||
|
||||
@ -65,9 +65,7 @@ class BFTNonValidatingNotaryService(override val services: ServiceHubInternal, c
|
||||
|
||||
fun commitTransaction(tx: Any, otherSide: Party) = client.commitTransaction(tx, otherSide)
|
||||
|
||||
override fun createServiceFlow(otherParty: Party, platformVersion: Int): FlowLogic<Void?> {
|
||||
return ServiceFlow(otherParty, this)
|
||||
}
|
||||
override fun createServiceFlow(otherParty: Party): FlowLogic<Void?> = ServiceFlow(otherParty, this)
|
||||
|
||||
private class ServiceFlow(val otherSide: Party, val service: BFTNonValidatingNotaryService) : FlowLogic<Void?>() {
|
||||
@Suspendable
|
||||
|
||||
@ -15,9 +15,7 @@ class RaftNonValidatingNotaryService(override val services: ServiceHubInternal)
|
||||
override val timeWindowChecker: TimeWindowChecker = TimeWindowChecker(services.clock)
|
||||
override val uniquenessProvider: RaftUniquenessProvider = RaftUniquenessProvider(services)
|
||||
|
||||
override fun createServiceFlow(otherParty: Party, platformVersion: Int): NotaryFlow.Service {
|
||||
return NonValidatingNotaryFlow(otherParty, this)
|
||||
}
|
||||
override fun createServiceFlow(otherParty: Party): NotaryFlow.Service = NonValidatingNotaryFlow(otherParty, this)
|
||||
|
||||
override fun start() {
|
||||
uniquenessProvider.start()
|
||||
|
||||
@ -15,9 +15,7 @@ class RaftValidatingNotaryService(override val services: ServiceHubInternal) : T
|
||||
override val timeWindowChecker: TimeWindowChecker = TimeWindowChecker(services.clock)
|
||||
override val uniquenessProvider: RaftUniquenessProvider = RaftUniquenessProvider(services)
|
||||
|
||||
override fun createServiceFlow(otherParty: Party, platformVersion: Int): NotaryFlow.Service {
|
||||
return ValidatingNotaryFlow(otherParty, this)
|
||||
}
|
||||
override fun createServiceFlow(otherParty: Party): NotaryFlow.Service = ValidatingNotaryFlow(otherParty, this)
|
||||
|
||||
override fun start() {
|
||||
uniquenessProvider.start()
|
||||
|
||||
@ -16,9 +16,7 @@ class SimpleNotaryService(override val services: ServiceHubInternal) : TrustedAu
|
||||
override val timeWindowChecker = TimeWindowChecker(services.clock)
|
||||
override val uniquenessProvider = PersistentUniquenessProvider()
|
||||
|
||||
override fun createServiceFlow(otherParty: Party, platformVersion: Int): NotaryFlow.Service {
|
||||
return NonValidatingNotaryFlow(otherParty, this)
|
||||
}
|
||||
override fun createServiceFlow(otherParty: Party): NotaryFlow.Service = NonValidatingNotaryFlow(otherParty, this)
|
||||
|
||||
override fun start() {}
|
||||
override fun stop() {}
|
||||
|
||||
@ -16,9 +16,7 @@ class ValidatingNotaryService(override val services: ServiceHubInternal) : Trust
|
||||
override val timeWindowChecker = TimeWindowChecker(services.clock)
|
||||
override val uniquenessProvider = PersistentUniquenessProvider()
|
||||
|
||||
override fun createServiceFlow(otherParty: Party, platformVersion: Int): NotaryFlow.Service {
|
||||
return ValidatingNotaryFlow(otherParty, this)
|
||||
}
|
||||
override fun createServiceFlow(otherParty: Party): NotaryFlow.Service = ValidatingNotaryFlow(otherParty, this)
|
||||
|
||||
override fun start() {}
|
||||
override fun stop() {}
|
||||
|
||||
@ -1,52 +0,0 @@
|
||||
package net.corda.node
|
||||
|
||||
import net.corda.core.internal.copyToDirectory
|
||||
import net.corda.core.internal.createDirectories
|
||||
import net.corda.core.internal.div
|
||||
import net.corda.nodeapi.User
|
||||
import net.corda.smoketesting.NodeConfig
|
||||
import net.corda.smoketesting.NodeProcess
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.bouncycastle.asn1.x500.X500Name
|
||||
import org.junit.Test
|
||||
import java.nio.file.Paths
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
class CordappScanningNodeProcessTest {
|
||||
private companion object {
|
||||
val user = User("user1", "test", permissions = setOf("ALL"))
|
||||
val port = AtomicInteger(15100)
|
||||
}
|
||||
|
||||
private val factory = NodeProcess.Factory()
|
||||
|
||||
private val aliceConfig = NodeConfig(
|
||||
legalName = X500Name("CN=Alice Corp,O=Alice Corp,L=Madrid,C=ES"),
|
||||
p2pPort = port.andIncrement,
|
||||
rpcPort = port.andIncrement,
|
||||
webPort = port.andIncrement,
|
||||
extraServices = emptyList(),
|
||||
users = listOf(user)
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `CorDapp jar in plugins directory is scanned`() {
|
||||
// If the CorDapp jar does't exist then run the smokeTestClasses gradle task
|
||||
val cordappJar = Paths.get(javaClass.getResource("/trader-demo.jar").toURI())
|
||||
val pluginsDir = (factory.baseDirectory(aliceConfig) / "plugins").createDirectories()
|
||||
cordappJar.copyToDirectory(pluginsDir)
|
||||
|
||||
factory.create(aliceConfig).use {
|
||||
it.connect().use {
|
||||
// If the CorDapp wasn't scanned then SellerFlow won't have been picked up as an RPC flow
|
||||
assertThat(it.proxy.registeredFlows()).contains("net.corda.traderdemo.flow.SellerFlow")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `empty plugins directory`() {
|
||||
(factory.baseDirectory(aliceConfig) / "plugins").createDirectories()
|
||||
factory.create(aliceConfig).close()
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,76 @@
|
||||
package net.corda.node
|
||||
|
||||
import co.paralleluniverse.fibers.Suspendable
|
||||
import net.corda.core.flows.*
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.copyToDirectory
|
||||
import net.corda.core.internal.createDirectories
|
||||
import net.corda.core.internal.div
|
||||
import net.corda.core.internal.list
|
||||
import net.corda.core.messaging.startFlow
|
||||
import net.corda.core.utilities.getOrThrow
|
||||
import net.corda.nodeapi.User
|
||||
import net.corda.smoketesting.NodeConfig
|
||||
import net.corda.smoketesting.NodeProcess
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.bouncycastle.asn1.x500.X500Name
|
||||
import org.junit.Test
|
||||
import java.nio.file.Paths
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.streams.toList
|
||||
|
||||
class CordappSmokeTest {
|
||||
private companion object {
|
||||
val user = User("user1", "test", permissions = setOf("ALL"))
|
||||
val port = AtomicInteger(15100)
|
||||
}
|
||||
|
||||
private val factory = NodeProcess.Factory()
|
||||
|
||||
private val aliceConfig = NodeConfig(
|
||||
legalName = X500Name("CN=Alice Corp,O=Alice Corp,L=Madrid,C=ES"),
|
||||
p2pPort = port.andIncrement,
|
||||
rpcPort = port.andIncrement,
|
||||
webPort = port.andIncrement,
|
||||
extraServices = emptyList(),
|
||||
users = listOf(user)
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `FlowContent appName returns the filename of the CorDapp jar`() {
|
||||
val pluginsDir = (factory.baseDirectory(aliceConfig) / "plugins").createDirectories()
|
||||
// Find the jar file for the smoke tests of this module
|
||||
val selfCorDapp = Paths.get("build", "libs").list {
|
||||
it.filter { "-smoke-test" in it.toString() }.toList().single()
|
||||
}
|
||||
selfCorDapp.copyToDirectory(pluginsDir)
|
||||
|
||||
factory.create(aliceConfig).use { alice ->
|
||||
alice.connect().use { connectionToAlice ->
|
||||
val aliceIdentity = connectionToAlice.proxy.nodeIdentity().legalIdentity
|
||||
val future = connectionToAlice.proxy.startFlow(::DummyInitiatingFlow, aliceIdentity).returnValue
|
||||
assertThat(future.getOrThrow().appName).isEqualTo(selfCorDapp.fileName.toString().removeSuffix(".jar"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `empty plugins directory`() {
|
||||
(factory.baseDirectory(aliceConfig) / "plugins").createDirectories()
|
||||
factory.create(aliceConfig).close()
|
||||
}
|
||||
|
||||
@InitiatingFlow
|
||||
@StartableByRPC
|
||||
class DummyInitiatingFlow(val otherParty: Party) : FlowLogic<FlowContext>() {
|
||||
@Suspendable
|
||||
override fun call() = getFlowContext(otherParty)
|
||||
}
|
||||
|
||||
@Suppress("unused")
|
||||
@InitiatedBy(DummyInitiatingFlow::class)
|
||||
class DummyInitiatedFlow(val otherParty: Party) : FlowLogic<Unit>() {
|
||||
@Suspendable
|
||||
override fun call() = Unit
|
||||
}
|
||||
}
|
||||
@ -4,18 +4,19 @@ import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
|
||||
import net.corda.core.concurrent.CordaFuture
|
||||
import net.corda.core.contracts.Amount
|
||||
import net.corda.core.crypto.SecureHash
|
||||
import net.corda.core.flows.FlowContext
|
||||
import net.corda.core.flows.FlowInitiator
|
||||
import net.corda.core.flows.FlowLogic
|
||||
import net.corda.core.internal.FlowStateMachine
|
||||
import net.corda.core.flows.StateMachineRunId
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.FlowStateMachine
|
||||
import net.corda.core.node.ServiceHub
|
||||
import net.corda.core.transactions.SignedTransaction
|
||||
import net.corda.testing.DUMMY_CA
|
||||
import net.corda.core.utilities.UntrustworthyData
|
||||
import net.corda.jackson.JacksonSupport
|
||||
import net.corda.node.services.identity.InMemoryIdentityService
|
||||
import net.corda.node.shell.InteractiveShell
|
||||
import net.corda.testing.DUMMY_CA
|
||||
import net.corda.testing.MEGA_CORP
|
||||
import net.corda.testing.MEGA_CORP_IDENTITY
|
||||
import org.junit.Test
|
||||
@ -70,32 +71,26 @@ class InteractiveShellTest {
|
||||
fun party() = check("party: \"${MEGA_CORP.name}\"", MEGA_CORP.name.toString())
|
||||
|
||||
class DummyFSM(val logic: FlowA) : FlowStateMachine<Any?> {
|
||||
override fun getFlowContext(otherParty: Party, sessionFlow: FlowLogic<*>): FlowContext {
|
||||
throw UnsupportedOperationException("not implemented")
|
||||
}
|
||||
override fun <T : Any> sendAndReceive(receiveType: Class<T>, otherParty: Party, payload: Any, sessionFlow: FlowLogic<*>, retrySend: Boolean): UntrustworthyData<T> {
|
||||
throw UnsupportedOperationException("not implemented")
|
||||
}
|
||||
|
||||
override fun <T : Any> receive(receiveType: Class<T>, otherParty: Party, sessionFlow: FlowLogic<*>): UntrustworthyData<T> {
|
||||
throw UnsupportedOperationException("not implemented")
|
||||
}
|
||||
|
||||
override fun send(otherParty: Party, payload: Any, sessionFlow: FlowLogic<*>) {
|
||||
throw UnsupportedOperationException("not implemented")
|
||||
}
|
||||
|
||||
override fun waitForLedgerCommit(hash: SecureHash, sessionFlow: FlowLogic<*>): SignedTransaction {
|
||||
throw UnsupportedOperationException("not implemented")
|
||||
}
|
||||
|
||||
override val serviceHub: ServiceHub
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val logger: Logger
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val id: StateMachineRunId
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val resultFuture: CordaFuture<Any?>
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val flowInitiator: FlowInitiator
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val serviceHub: ServiceHub get() = throw UnsupportedOperationException()
|
||||
override val logger: Logger get() = throw UnsupportedOperationException()
|
||||
override val id: StateMachineRunId get() = throw UnsupportedOperationException()
|
||||
override val resultFuture: CordaFuture<Any?> get() = throw UnsupportedOperationException()
|
||||
override val flowInitiator: FlowInitiator get() = throw UnsupportedOperationException()
|
||||
|
||||
override fun checkFlowPermission(permissionName: String, extraAuditData: Map<String, String>) {
|
||||
// Do nothing
|
||||
|
||||
@ -260,15 +260,15 @@ class FlowFrameworkTests {
|
||||
assertThat(node3Flow.receivedPayloads[0]).isEqualTo(payload)
|
||||
|
||||
assertSessionTransfers(node2,
|
||||
node1 sent sessionInit(SendFlow::class, 1, payload) to node2,
|
||||
node2 sent sessionConfirm to node1,
|
||||
node1 sent sessionInit(SendFlow::class, payload = payload) to node2,
|
||||
node2 sent sessionConfirm() to node1,
|
||||
node1 sent normalEnd to node2
|
||||
//There's no session end from the other flows as they're manually suspended
|
||||
)
|
||||
|
||||
assertSessionTransfers(node3,
|
||||
node1 sent sessionInit(SendFlow::class, 1, payload) to node3,
|
||||
node3 sent sessionConfirm to node1,
|
||||
node1 sent sessionInit(SendFlow::class, payload = payload) to node3,
|
||||
node3 sent sessionConfirm() to node1,
|
||||
node1 sent normalEnd to node3
|
||||
//There's no session end from the other flows as they're manually suspended
|
||||
)
|
||||
@ -294,14 +294,14 @@ class FlowFrameworkTests {
|
||||
|
||||
assertSessionTransfers(node2,
|
||||
node1 sent sessionInit(ReceiveFlow::class) to node2,
|
||||
node2 sent sessionConfirm to node1,
|
||||
node2 sent sessionConfirm() to node1,
|
||||
node2 sent sessionData(node2Payload) to node1,
|
||||
node2 sent normalEnd to node1
|
||||
)
|
||||
|
||||
assertSessionTransfers(node3,
|
||||
node1 sent sessionInit(ReceiveFlow::class) to node3,
|
||||
node3 sent sessionConfirm to node1,
|
||||
node3 sent sessionConfirm() to node1,
|
||||
node3 sent sessionData(node3Payload) to node1,
|
||||
node3 sent normalEnd to node1
|
||||
)
|
||||
@ -314,8 +314,8 @@ class FlowFrameworkTests {
|
||||
mockNet.runNetwork()
|
||||
|
||||
assertSessionTransfers(
|
||||
node1 sent sessionInit(PingPongFlow::class, 1, 10L) to node2,
|
||||
node2 sent sessionConfirm to node1,
|
||||
node1 sent sessionInit(PingPongFlow::class, payload = 10L) to node2,
|
||||
node2 sent sessionConfirm() to node1,
|
||||
node2 sent sessionData(20L) to node1,
|
||||
node1 sent sessionData(11L) to node2,
|
||||
node2 sent sessionData(21L) to node1,
|
||||
@ -419,7 +419,7 @@ class FlowFrameworkTests {
|
||||
|
||||
assertSessionTransfers(
|
||||
node1 sent sessionInit(ReceiveFlow::class) to node2,
|
||||
node2 sent sessionConfirm to node1,
|
||||
node2 sent sessionConfirm() to node1,
|
||||
node2 sent erroredEnd() to node1
|
||||
)
|
||||
}
|
||||
@ -452,7 +452,7 @@ class FlowFrameworkTests {
|
||||
|
||||
assertSessionTransfers(
|
||||
node1 sent sessionInit(ReceiveFlow::class) to node2,
|
||||
node2 sent sessionConfirm to node1,
|
||||
node2 sent sessionConfirm() to node1,
|
||||
node2 sent erroredEnd(erroringFlow.get().exceptionThrown) to node1
|
||||
)
|
||||
// Make sure the original stack trace isn't sent down the wire
|
||||
@ -500,7 +500,7 @@ class FlowFrameworkTests {
|
||||
|
||||
assertSessionTransfers(node2,
|
||||
node1 sent sessionInit(ReceiveFlow::class) to node2,
|
||||
node2 sent sessionConfirm to node1,
|
||||
node2 sent sessionConfirm() to node1,
|
||||
node2 sent sessionData("Hello") to node1,
|
||||
node1 sent erroredEnd() to node2
|
||||
)
|
||||
@ -627,26 +627,30 @@ class FlowFrameworkTests {
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `upgraded flow`() {
|
||||
node1.services.startFlow(UpgradedFlow(node2.info.legalIdentity))
|
||||
fun `upgraded initiating flow`() {
|
||||
node2.registerFlowFactory(UpgradedFlow::class, initiatedFlowVersion = 1) { SendFlow("Old initiated", it) }
|
||||
val result = node1.services.startFlow(UpgradedFlow(node2.info.legalIdentity)).resultFuture
|
||||
mockNet.runNetwork()
|
||||
assertThat(sessionTransfers).startsWith(
|
||||
node1 sent sessionInit(UpgradedFlow::class, 2) to node2
|
||||
node1 sent sessionInit(UpgradedFlow::class, flowVersion = 2) to node2,
|
||||
node2 sent sessionConfirm(flowVersion = 1) to node1
|
||||
)
|
||||
val (receivedPayload, node2FlowVersion) = result.getOrThrow()
|
||||
assertThat(receivedPayload).isEqualTo("Old initiated")
|
||||
assertThat(node2FlowVersion).isEqualTo(1)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `unsupported new flow version`() {
|
||||
node2.internalRegisterFlowFactory(
|
||||
UpgradedFlow::class.java,
|
||||
InitiatedFlowFactory.CorDapp(version = 1, factory = ::DoubleInlinedSubFlow),
|
||||
DoubleInlinedSubFlow::class.java,
|
||||
track = false)
|
||||
val result = node1.services.startFlow(UpgradedFlow(node2.info.legalIdentity)).resultFuture
|
||||
fun `upgraded initiated flow`() {
|
||||
node2.registerFlowFactory(SendFlow::class, initiatedFlowVersion = 2) { UpgradedFlow(it) }
|
||||
val initiatingFlow = SendFlow("Old initiating", node2.info.legalIdentity)
|
||||
node1.services.startFlow(initiatingFlow)
|
||||
mockNet.runNetwork()
|
||||
assertThatExceptionOfType(UnexpectedFlowEndException::class.java)
|
||||
.isThrownBy { result.getOrThrow() }
|
||||
.withMessageContaining("Version")
|
||||
assertThat(sessionTransfers).startsWith(
|
||||
node1 sent sessionInit(SendFlow::class, flowVersion = 1, payload = "Old initiating") to node2,
|
||||
node2 sent sessionConfirm(flowVersion = 2) to node1
|
||||
)
|
||||
assertThat(initiatingFlow.getFlowContext(node2.info.legalIdentity).flowVersion).isEqualTo(2)
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -660,7 +664,7 @@ class FlowFrameworkTests {
|
||||
|
||||
@Test
|
||||
fun `unknown class in session init`() {
|
||||
node1.sendSessionMessage(SessionInit(random63BitValue(), "not.a.real.Class", 1, null), node2)
|
||||
node1.sendSessionMessage(SessionInit(random63BitValue(), "not.a.real.Class", 1, "version", null), node2)
|
||||
mockNet.runNetwork()
|
||||
assertThat(sessionTransfers).hasSize(2) // Only the session-init and session-reject are expected
|
||||
val reject = sessionTransfers.last().message as SessionReject
|
||||
@ -669,7 +673,7 @@ class FlowFrameworkTests {
|
||||
|
||||
@Test
|
||||
fun `non-flow class in session init`() {
|
||||
node1.sendSessionMessage(SessionInit(random63BitValue(), String::class.java.name, 1, null), node2)
|
||||
node1.sendSessionMessage(SessionInit(random63BitValue(), String::class.java.name, 1, "version", null), node2)
|
||||
mockNet.runNetwork()
|
||||
assertThat(sessionTransfers).hasSize(2) // Only the session-init and session-reject are expected
|
||||
val reject = sessionTransfers.last().message as SessionReject
|
||||
@ -678,7 +682,7 @@ class FlowFrameworkTests {
|
||||
|
||||
@Test
|
||||
fun `single inlined sub-flow`() {
|
||||
node2.registerFlowFactory(SendAndReceiveFlow::class, ::SingleInlinedSubFlow)
|
||||
node2.registerFlowFactory(SendAndReceiveFlow::class) { SingleInlinedSubFlow(it) }
|
||||
val result = node1.services.startFlow(SendAndReceiveFlow(node2.info.legalIdentity, "Hello")).resultFuture
|
||||
mockNet.runNetwork()
|
||||
assertThat(result.getOrThrow()).isEqualTo("HelloHello")
|
||||
@ -686,7 +690,7 @@ class FlowFrameworkTests {
|
||||
|
||||
@Test
|
||||
fun `double inlined sub-flow`() {
|
||||
node2.registerFlowFactory(SendAndReceiveFlow::class, ::DoubleInlinedSubFlow)
|
||||
node2.registerFlowFactory(SendAndReceiveFlow::class) { DoubleInlinedSubFlow(it) }
|
||||
val result = node1.services.startFlow(SendAndReceiveFlow(node2.info.legalIdentity, "Hello")).resultFuture
|
||||
mockNet.runNetwork()
|
||||
assertThat(result.getOrThrow()).isEqualTo("HelloHello")
|
||||
@ -711,21 +715,22 @@ class FlowFrameworkTests {
|
||||
}
|
||||
|
||||
private inline fun <reified P : FlowLogic<*>> MockNode.registerFlowFactory(
|
||||
initiatingFlowClass: KClass<out FlowLogic<*>>,
|
||||
noinline flowFactory: (Party) -> P): CordaFuture<P>
|
||||
initiatingFlowClass: KClass<out FlowLogic<*>>,
|
||||
initiatedFlowVersion: Int = 1,
|
||||
noinline flowFactory: (Party) -> P): CordaFuture<P>
|
||||
{
|
||||
val observable = internalRegisterFlowFactory(initiatingFlowClass.java, object : InitiatedFlowFactory<P> {
|
||||
override fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): P {
|
||||
return flowFactory(otherParty)
|
||||
}
|
||||
}, P::class.java, track = true)
|
||||
val observable = internalRegisterFlowFactory(
|
||||
initiatingFlowClass.java,
|
||||
InitiatedFlowFactory.CorDapp(initiatedFlowVersion, "", flowFactory),
|
||||
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.name, flowVersion, payload)
|
||||
return SessionInit(0, clientFlowClass.java.name, flowVersion, "", payload)
|
||||
}
|
||||
private val sessionConfirm = SessionConfirm(0, 0)
|
||||
private fun sessionConfirm(flowVersion: Int = 1) = SessionConfirm(0, 0, flowVersion, "")
|
||||
private fun sessionData(payload: Any) = SessionData(0, payload)
|
||||
private val normalEnd = NormalSessionEnd(0)
|
||||
private fun erroredEnd(errorResponse: FlowException? = null) = ErrorSessionEnd(0, errorResponse)
|
||||
@ -762,8 +767,8 @@ class FlowFrameworkTests {
|
||||
|
||||
private fun sanitise(message: SessionMessage) = when (message) {
|
||||
is SessionData -> message.copy(recipientSessionId = 0)
|
||||
is SessionInit -> message.copy(initiatorSessionId = 0)
|
||||
is SessionConfirm -> message.copy(initiatorSessionId = 0, initiatedSessionId = 0)
|
||||
is SessionInit -> message.copy(initiatorSessionId = 0, appName = "")
|
||||
is SessionConfirm -> message.copy(initiatorSessionId = 0, initiatedSessionId = 0, appName = "")
|
||||
is NormalSessionEnd -> message.copy(recipientSessionId = 0)
|
||||
is ErrorSessionEnd -> message.copy(recipientSessionId = 0)
|
||||
else -> message
|
||||
@ -799,7 +804,6 @@ class FlowFrameworkTests {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@InitiatingFlow
|
||||
private open class SendFlow(val payload: Any, vararg val otherParties: Party) : FlowLogic<Unit>() {
|
||||
init {
|
||||
@ -921,9 +925,13 @@ class FlowFrameworkTests {
|
||||
}
|
||||
|
||||
@InitiatingFlow(version = 2)
|
||||
private class UpgradedFlow(val otherParty: Party) : FlowLogic<Any>() {
|
||||
private class UpgradedFlow(val otherParty: Party) : FlowLogic<Pair<Any, Int>>() {
|
||||
@Suspendable
|
||||
override fun call(): Any = receive<Any>(otherParty).unwrap { it }
|
||||
override fun call(): Pair<Any, Int> {
|
||||
val received = receive<Any>(otherParty).unwrap { it }
|
||||
val otherFlowVersion = getFlowContext(otherParty).flowVersion
|
||||
return Pair(received, otherFlowVersion)
|
||||
}
|
||||
}
|
||||
|
||||
private class SingleInlinedSubFlow(val otherParty: Party) : FlowLogic<Unit>() {
|
||||
|
||||
Reference in New Issue
Block a user