ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-03-10 14:34:14 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Changed the counter parties in the query to be CordaX509Name (#6622)

Browse Source
This commit is contained in:
Alexey Kadyrov 2020-08-12 14:53:03 +01:00 committed by GitHub
parent 57de0c4eec
commit 518026c6c4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 46 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
node/src/main/kotlin/net/corda/node/services/statemachine/FlowOperator.kt
View File

@ -1,6 +1,7 @@
package net.corda.node.services.statemachine package net.corda.node.services.statemachine
import net.corda.core.flows.StateMachineRunId import net.corda.core.flows.StateMachineRunId
import net.corda.core.identity.CordaX500Name
import net.corda.core.identity.Party import net.corda.core.identity.Party
import net.corda.core.internal.FlowIORequest import net.corda.core.internal.FlowIORequest
import net.corda.core.internal.FlowStateMachine import net.corda.core.internal.FlowStateMachine
@ -10,18 +11,18 @@ import java.time.Duration
import java.time.Instant import java.time.Instant
/** /**
* Stage in which the flow is suspended * Defines criteria to get waiting flows
*/ */
enum class WaitingSource { data class WaitingFlowQuery(
SEND, val flowIds: MutableList<StateMachineRunId> = mutableListOf(),
RECEIVE, val onlyIfSuspendedLongerThan: Duration = Duration.ZERO,
SEND_AND_RECEIVE, val waitingSources: MutableList<WaitingSource> = mutableListOf(),
CLOSE_SESSIONS, val counterParties: MutableList<CordaX500Name> = mutableListOf()
WAIT_FOR_LEDGER_COMMIT, ) {
GET_FLOW_INFO, fun isDefined() = flowIds.isNotEmpty()
SLEEP, || waitingSources.isNotEmpty()
WAIT_FOR_SESSIONS_CONFIRMATIONS, || counterParties.isNotEmpty()
EXTERNAL_OPERATION || onlyIfSuspendedLongerThan > Duration.ZERO
} }
/** /**
@ -40,18 +41,18 @@ data class WaitingFlowInfo(
) )
/** /**
* Defines criteria to get waiting flows * Stage in which the flow is suspended
*/ */
data class WaitingFlowQuery( enum class WaitingSource {
val ids: MutableList<StateMachineRunId> = mutableListOf(), SEND,
val onlyIfSuspendedLongerThan: Duration = Duration.ZERO, RECEIVE,
val waitingSource: MutableList<WaitingSource> = mutableListOf(), SEND_AND_RECEIVE,
val counterParties: MutableList<Party> = mutableListOf() CLOSE_SESSIONS,
) { WAIT_FOR_LEDGER_COMMIT,
fun isDefined() = ids.isNotEmpty() GET_FLOW_INFO,
|| waitingSource.isNotEmpty() SLEEP,
|| counterParties.isNotEmpty() WAIT_FOR_SESSIONS_CONFIRMATIONS,
|| onlyIfSuspendedLongerThan > Duration.ZERO EXTERNAL_OPERATION
} }
/** /**
@ -89,14 +90,14 @@ class FlowOperator(private val smm: StateMachineManager, private val clock: Cloc
override fun queryWaitingFlows(query: WaitingFlowQuery): Set<WaitingFlowInfo> { override fun queryWaitingFlows(query: WaitingFlowQuery): Set<WaitingFlowInfo> {
var sequence = getAllWaitingFlows() var sequence = getAllWaitingFlows()
if (query.ids.isNotEmpty()) { if (query.flowIds.isNotEmpty()) {
sequence = sequence.filter { it.id in query.ids } sequence = sequence.filter { it.id in query.flowIds }
} }
if (query.counterParties.isNotEmpty()) { if (query.counterParties.isNotEmpty()) {
sequence = sequence.filter { it.isWaitingForParties(query.counterParties) } sequence = sequence.filter { it.isWaitingForParties(query.counterParties) }
} }
if (query.waitingSource.isNotEmpty()) { if (query.waitingSources.isNotEmpty()) {
sequence = sequence.filter { it.waitingSource() in query.waitingSource } sequence = sequence.filter { it.waitingSource() in query.waitingSources }
} }
if (query.onlyIfSuspendedLongerThan > Duration.ZERO) { if (query.onlyIfSuspendedLongerThan > Duration.ZERO) {
val now = clock.instant() val now = clock.instant()
@ -125,13 +126,13 @@ class FlowOperator(private val smm: StateMachineManager, private val clock: Cloc
} }
} }
private fun FlowStateMachineImpl<*>.isWaitingForParties(parties: List<Party>): Boolean { private fun FlowStateMachineImpl<*>.isWaitingForParties(parties: List<CordaX500Name>): Boolean {
return ioRequest()?.let { request -> return ioRequest()?.let { request ->
when (request) { when (request) {
is FlowIORequest.GetFlowInfo -> request.sessions.any { it.counterparty in parties } is FlowIORequest.GetFlowInfo -> request.sessions.any { it.counterparty.name in parties }
is FlowIORequest.Receive -> request.sessions.any { it.counterparty in parties } is FlowIORequest.Receive -> request.sessions.any { it.counterparty.name in parties }
is FlowIORequest.Send -> request.sessionToMessage.keys.any { it.counterparty in parties } is FlowIORequest.Send -> request.sessionToMessage.keys.any { it.counterparty.name in parties }
is FlowIORequest.SendAndReceive -> request.sessionToMessage.keys.any { it.counterparty in parties } is FlowIORequest.SendAndReceive -> request.sessionToMessage.keys.any { it.counterparty.name in parties }
else -> false else -> false
} }
} ?: false } ?: false

28
node/src/test/kotlin/net/corda/node/services/statemachine/FlowOperatorTests.kt
View File

@ -104,7 +104,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty) WaitingFlowQuery(counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME)
)) ))
assertEquals(2, result.size) assertEquals(2, result.size)
@ -132,7 +132,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(daveParty) WaitingFlowQuery(counterParties = mutableListOf(DAVE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -152,7 +152,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty) WaitingFlowQuery(counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -176,7 +176,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(bobParty, daveParty) WaitingFlowQuery(counterParties = mutableListOf(BOB_NAME, DAVE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -199,7 +199,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery( WaitingFlowQuery(
counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty), counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME),
onlyIfSuspendedLongerThan = 4.seconds onlyIfSuspendedLongerThan = 4.seconds
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -227,8 +227,8 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery( WaitingFlowQuery(
counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty), counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME),
waitingSource = mutableListOf(WaitingSource.EXTERNAL_OPERATION, WaitingSource.RECEIVE, WaitingSource.GET_FLOW_INFO) waitingSources = mutableListOf(WaitingSource.EXTERNAL_OPERATION, WaitingSource.RECEIVE, WaitingSource.GET_FLOW_INFO)
)) ))
assertEquals(2, result.size) assertEquals(2, result.size)
@ -258,7 +258,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryFlowsCurrentlyWaitingForPartiesGrouped(WaitingFlowQuery( val result = cut.queryFlowsCurrentlyWaitingForPartiesGrouped(WaitingFlowQuery(
waitingSource = mutableListOf(WaitingSource.EXTERNAL_OPERATION, WaitingSource.RECEIVE) waitingSources = mutableListOf(WaitingSource.EXTERNAL_OPERATION, WaitingSource.RECEIVE)
)) ))
assertEquals(2, result.size) assertEquals(2, result.size)
@ -303,7 +303,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(ids = mutableListOf(charlieStart.id, eugeneStart.id) WaitingFlowQuery(flowIds = mutableListOf(charlieStart.id, eugeneStart.id)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -323,7 +323,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty) WaitingFlowQuery(counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -343,7 +343,7 @@ class FlowOperatorTests {
executeTest(5.seconds) { executeTest(5.seconds) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty) WaitingFlowQuery(counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -366,7 +366,7 @@ class FlowOperatorTests {
executeTest(5.seconds, { future.complete(Unit) }) { executeTest(5.seconds, { future.complete(Unit) }) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty) WaitingFlowQuery(counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)
@ -387,7 +387,7 @@ class FlowOperatorTests {
executeTest(5.seconds, { future.complete("Hello") }) { executeTest(5.seconds, { future.complete("Hello") }) {
val result = cut.queryWaitingFlows(WaitingFlowQuery( val result = cut.queryWaitingFlows(WaitingFlowQuery(
waitingSource = mutableListOf(WaitingSource.EXTERNAL_OPERATION) waitingSources = mutableListOf(WaitingSource.EXTERNAL_OPERATION)
)) // the list of counter parties must be empty to get any external operation )) // the list of counter parties must be empty to get any external operation
assertEquals(1, result.size) assertEquals(1, result.size)
@ -444,7 +444,7 @@ class FlowOperatorTests {
executeTest(5.seconds, { future.complete(Unit) }) { executeTest(5.seconds, { future.complete(Unit) }) {
val result = cut.queryWaitingFlows( val result = cut.queryWaitingFlows(
WaitingFlowQuery(counterParties = mutableListOf(aliceParty, bobParty, charlieParty, daveParty, eugeneParty) WaitingFlowQuery(counterParties = mutableListOf(ALICE_NAME, BOB_NAME, CHARLIE_NAME, DAVE_NAME, EUGENE_NAME)
)) ))
assertEquals(1, result.size) assertEquals(1, result.size)