ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-06-16 22:28:15 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

StateMachineManager is no longer lateinit. (#2123)

Browse Source
This commit is contained in:
Andrzej Cichocki
2017-11-27 17:55:08 +00:00
committed by GitHub
parent 4ca54b73fe
commit 4bd6fef0f9
20 changed files with 63 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -213,7 +213,7 @@ abstract class MQSecurityTest : NodeBasedTest() {
private fun startBobAndCommunicateWithAlice(): Party {
val bob = startNode(BOB.name)
bob.internals.registerInitiatedFlow(ReceiveFlow::class.java)
bob.registerInitiatedFlow(ReceiveFlow::class.java)
val bobParty = bob.info.chooseIdentity()
// Perform a protocol exchange to force the peer queue to be created
alice.services.startFlow(SendFlow(bobParty, 0)).resultFuture.getOrThrow()

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

@ -124,7 +124,6 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
private lateinit var _services: ServiceHubInternalImpl
protected var myNotaryIdentity: PartyAndCertificate? = null
protected lateinit var checkpointStorage: CheckpointStorage
protected lateinit var smm: StateMachineManager
private lateinit var tokenizableServices: List<Any>
protected lateinit var attachments: NodeAttachmentService
protected lateinit var network: MessagingService
@ -153,7 +152,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
@Volatile private var _started: StartedNode<AbstractNode>? = null
/** The implementation of the [CordaRPCOps] interface used by this node. */
open fun makeRPCOps(flowStarter: FlowStarter, database: CordaPersistence): CordaRPCOps {
open fun makeRPCOps(flowStarter: FlowStarter, database: CordaPersistence, smm: StateMachineManager): CordaRPCOps {
return SecureCordaRPCOps(services, smm, database, flowStarter)
}
@ -191,7 +190,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
val stateLoader = StateLoaderImpl(transactionStorage)
val nodeServices = makeServices(keyPairs, schemaService, transactionStorage, stateLoader, database, info, identityService)
val notaryService = makeNotaryService(nodeServices, database)
smm = makeStateMachineManager(database)
val smm = makeStateMachineManager(database)
val flowStarter = FlowStarterImpl(serverThread, smm)
val schedulerService = NodeSchedulerService(
platformClock,
@ -208,13 +207,13 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
MoreExecutors.shutdownAndAwaitTermination(serverThread as ExecutorService, 50, SECONDS)
}
}
makeVaultObservers(schedulerService, database.hibernateConfig)
val rpcOps = makeRPCOps(flowStarter, database)
makeVaultObservers(schedulerService, database.hibernateConfig, smm)
val rpcOps = makeRPCOps(flowStarter, database, smm)
startMessagingService(rpcOps)
installCoreFlows()
val cordaServices = installCordaServices(flowStarter)
tokenizableServices = nodeServices + cordaServices + schedulerService
registerCordappFlows()
registerCordappFlows(smm)
_services.rpcFlows += cordappLoader.cordapps.flatMap { it.rpcFlows }
FlowLogicRefFactoryImpl.classloader = cordappLoader.appClassLoader
startShell(rpcOps)
@ -397,11 +396,11 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
installCoreFlow(NotaryFlow.Client::class, service::createServiceFlow)
}
private fun registerCordappFlows() {
private fun registerCordappFlows(smm: StateMachineManager) {
cordappLoader.cordapps.flatMap { it.initiatedFlows }
.forEach {
try {
registerInitiatedFlowInternal(it, track = false)
registerInitiatedFlowInternal(smm, 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}")
@ -411,13 +410,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
}
}
/**
* 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)
internal fun <T : FlowLogic<*>> registerInitiatedFlow(smm: StateMachineManager, initiatedFlowClass: Class<T>): Observable<T> {
return registerInitiatedFlowInternal(smm, initiatedFlowClass, track = true)
}
// TODO remove once not needed
@ -426,7 +420,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
"It should accept a ${FlowSession::class.java.simpleName} instead"
}
private fun <F : FlowLogic<*>> registerInitiatedFlowInternal(initiatedFlow: Class<F>, track: Boolean): Observable<F> {
private fun <F : FlowLogic<*>> registerInitiatedFlowInternal(smm: StateMachineManager, initiatedFlow: Class<F>, track: Boolean): Observable<F> {
val constructors = initiatedFlow.declaredConstructors.associateBy { it.parameterTypes.toList() }
val flowSessionCtor = constructors[listOf(FlowSession::class.java)]?.apply { isAccessible = true }
val ctor: (FlowSession) -> F = if (flowSessionCtor == null) {
@ -447,16 +441,16 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
"${InitiatedBy::class.java.name} must point to ${classWithAnnotation.name} and not ${initiatingFlow.name}"
}
val flowFactory = InitiatedFlowFactory.CorDapp(version, initiatedFlow.appName, ctor)
val observable = internalRegisterFlowFactory(initiatingFlow, flowFactory, initiatedFlow, track)
val observable = internalRegisterFlowFactory(smm, initiatingFlow, flowFactory, initiatedFlow, track)
log.info("Registered ${initiatingFlow.name} to initiate ${initiatedFlow.name} (version $version)")
return observable
}
@VisibleForTesting
fun <F : FlowLogic<*>> internalRegisterFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>,
flowFactory: InitiatedFlowFactory<F>,
initiatedFlowClass: Class<F>,
track: Boolean): Observable<F> {
internal fun <F : FlowLogic<*>> internalRegisterFlowFactory(smm: StateMachineManager,
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 {
@ -519,7 +513,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
}
protected open fun makeTransactionStorage(database: CordaPersistence): WritableTransactionStorage = DBTransactionStorage()
private fun makeVaultObservers(schedulerService: SchedulerService, hibernateConfig: HibernateConfiguration) {
private fun makeVaultObservers(schedulerService: SchedulerService, hibernateConfig: HibernateConfiguration, smm: StateMachineManager) {
VaultSoftLockManager.install(services.vaultService, smm)
ScheduledActivityObserver.install(services.vaultService, schedulerService)
HibernateObserver.install(services.vaultService.rawUpdates, hibernateConfig)

18
node/src/main/kotlin/net/corda/node/internal/StartedNode.kt
View File

@ -2,6 +2,8 @@ package net.corda.node.internal
import net.corda.core.contracts.*
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
import net.corda.core.internal.VisibleForTesting
import net.corda.core.messaging.CordaRPCOps
import net.corda.core.node.NodeInfo
import net.corda.core.node.StateLoader
@ -13,6 +15,7 @@ import net.corda.node.services.messaging.MessagingService
import net.corda.node.services.persistence.NodeAttachmentService
import net.corda.node.services.statemachine.StateMachineManager
import net.corda.node.utilities.CordaPersistence
import rx.Observable
interface StartedNode<out N : AbstractNode> {
val internals: N
@ -26,7 +29,20 @@ interface StartedNode<out N : AbstractNode> {
val rpcOps: CordaRPCOps
val notaryService: NotaryService?
fun dispose() = internals.stop()
fun <T : FlowLogic<*>> registerInitiatedFlow(initiatedFlowClass: Class<T>) = internals.registerInitiatedFlow(initiatedFlowClass)
/**
* 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>) = internals.registerInitiatedFlow(smm, initiatedFlowClass)
@VisibleForTesting
fun <F : FlowLogic<*>> internalRegisterFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>,
flowFactory: InitiatedFlowFactory<F>,
initiatedFlowClass: Class<F>,
track: Boolean): Observable<F> {
return internals.internalRegisterFlowFactory(smm, initiatingFlowClass, flowFactory, initiatedFlowClass, track)
}
}
class StateLoaderImpl(private val validatedTransactions: TransactionStorage) : StateLoader {

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

@ -519,7 +519,7 @@ class TwoPartyTradeFlowTests(private val anonymous: Boolean) {
sellerNode: StartedNode<MockNetwork.MockNode>,
buyerNode: StartedNode<MockNetwork.MockNode>,
assetToSell: StateAndRef<OwnableState>): RunResult {
val buyerFlows: Observable<out FlowLogic<*>> = buyerNode.internals.registerInitiatedFlow(BuyerAcceptor::class.java)
val buyerFlows: Observable<out FlowLogic<*>> = buyerNode.registerInitiatedFlow(BuyerAcceptor::class.java)
val firstBuyerFiber = buyerFlows.toFuture().map { it.stateMachine }
val seller = SellerInitiator(buyer, notary, assetToSell, 1000.DOLLARS, anonymous)
val sellerResult = sellerNode.services.startFlow(seller).resultFuture

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

@ -672,7 +672,7 @@ class FlowFrameworkTests {
initiatingFlowClass: KClass<out FlowLogic<*>>,
initiatedFlowVersion: Int = 1,
noinline flowFactory: (FlowSession) -> P): CordaFuture<P> {
val observable = internals.internalRegisterFlowFactory(
val observable = internalRegisterFlowFactory(
initiatingFlowClass.java,
InitiatedFlowFactory.CorDapp(initiatedFlowVersion, "", flowFactory),
P::class.java,

2
node/src/test/kotlin/net/corda/node/services/vault/VaultSoftLockManagerTest.kt
View File

@ -66,7 +66,7 @@ class NodePair(private val mockNet: MockNetwork) {
private set
fun <T> communicate(clientLogic: AbstractClientLogic<T>, rebootClient: Boolean): FlowStateMachine<T> {
server.internals.internalRegisterFlowFactory(AbstractClientLogic::class.java, InitiatedFlowFactory.Core { ServerLogic(it, serverRunning) }, ServerLogic::class.java, false)
server.internalRegisterFlowFactory(AbstractClientLogic::class.java, InitiatedFlowFactory.Core { ServerLogic(it, serverRunning) }, ServerLogic::class.java, false)
client.services.startFlow(clientLogic)
while (!serverRunning.get()) mockNet.runNetwork(1)
if (rebootClient) {