From ae85759067c15bdc2c7e4856782e58e99518b0f4 Mon Sep 17 00:00:00 2001 From: Patrick Kuo Date: Fri, 18 Aug 2017 09:46:59 +0100 Subject: [PATCH] Added notifyVault flag to recordTransaction and skip notify for irrelevant TX (#1250) * Add notifyVault flag to recordTransaction to skip notifying the vault for transactions from ResolveTransactionFlow. * added methods for use in Java * reverted format changes * addressed PR issues changed recordTransaction method signature --- .../core/internal/ResolveTransactionsFlow.kt | 2 +- .../kotlin/net/corda/core/node/ServiceHub.kt | 29 ++++++++++++++----- .../net/corda/node/internal/AbstractNode.kt | 4 +-- .../node/services/api/ServiceHubInternal.kt | 10 ++++--- .../services/vault/HibernateVaultQueryImpl.kt | 1 - .../database/HibernateConfigurationTest.kt | 2 +- .../services/vault/NodeVaultServiceTest.kt | 5 ++-- .../net/corda/testing/node/MockServices.kt | 4 +-- 8 files changed, 37 insertions(+), 20 deletions(-) diff --git a/core/src/main/kotlin/net/corda/core/internal/ResolveTransactionsFlow.kt b/core/src/main/kotlin/net/corda/core/internal/ResolveTransactionsFlow.kt index 77fa327e91..e776b0efce 100644 --- a/core/src/main/kotlin/net/corda/core/internal/ResolveTransactionsFlow.kt +++ b/core/src/main/kotlin/net/corda/core/internal/ResolveTransactionsFlow.kt @@ -92,7 +92,7 @@ class ResolveTransactionsFlow(private val txHashes: Set, // half way through, it's no big deal, although it might result in us attempting to re-download data // redundantly next time we attempt verification. it.verify(serviceHub) - serviceHub.recordTransactions(it) + serviceHub.recordTransactions(false, it) } return signedTransaction?.let { diff --git a/core/src/main/kotlin/net/corda/core/node/ServiceHub.kt b/core/src/main/kotlin/net/corda/core/node/ServiceHub.kt index 0a885130d9..fb882eb4ad 100644 --- a/core/src/main/kotlin/net/corda/core/node/ServiceHub.kt +++ b/core/src/main/kotlin/net/corda/core/node/ServiceHub.kt @@ -68,18 +68,35 @@ interface ServiceHub : ServicesForResolution { /** * Stores the given [SignedTransaction]s in the local transaction storage and then sends them to the vault for - * further processing. This is expected to be run within a database transaction. + * further processing if [notifyVault] is true. This is expected to be run within a database transaction. * * @param txs The transactions to record. + * @param notifyVault indicate if the vault should be notified for the update. */ - fun recordTransactions(txs: Iterable) + fun recordTransactions(notifyVault: Boolean, txs: Iterable) + + /** + * Stores the given [SignedTransaction]s in the local transaction storage and then sends them to the vault for + * further processing if [notifyVault] is true. This is expected to be run within a database transaction. + */ + fun recordTransactions(notifyVault: Boolean, first: SignedTransaction, vararg remaining: SignedTransaction) { + recordTransactions(notifyVault, listOf(first, *remaining)) + } /** * Stores the given [SignedTransaction]s in the local transaction storage and then sends them to the vault for * further processing. This is expected to be run within a database transaction. */ fun recordTransactions(first: SignedTransaction, vararg remaining: SignedTransaction) { - recordTransactions(listOf(first, *remaining)) + recordTransactions(true, first, *remaining) + } + + /** + * Stores the given [SignedTransaction]s in the local transaction storage and then sends them to the vault for + * further processing. This is expected to be run within a database transaction. + */ + fun recordTransactions(txs: Iterable) { + recordTransactions(true, txs) } /** @@ -92,8 +109,7 @@ interface ServiceHub : ServicesForResolution { val stx = validatedTransactions.getTransaction(stateRef.txhash) ?: throw TransactionResolutionException(stateRef.txhash) return if (stx.isNotaryChangeTransaction()) { stx.resolveNotaryChangeTransaction(this).outputs[stateRef.index] - } - else stx.tx.outputs[stateRef.index] + } else stx.tx.outputs[stateRef.index] } /** @@ -106,8 +122,7 @@ interface ServiceHub : ServicesForResolution { val stx = validatedTransactions.getTransaction(stateRef.txhash) ?: throw TransactionResolutionException(stateRef.txhash) return if (stx.isNotaryChangeTransaction()) { stx.resolveNotaryChangeTransaction(this).outRef(stateRef.index) - } - else { + } else { stx.tx.outRef(stateRef.index) } } diff --git a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt index 84da8e024c..4b5db6c46a 100644 --- a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt +++ b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt @@ -801,9 +801,9 @@ abstract class AbstractNode(open val configuration: NodeConfiguration, return flowFactories[initiatingFlowClass] } - override fun recordTransactions(txs: Iterable) { + override fun recordTransactions(notifyVault: Boolean, txs: Iterable) { database.transaction { - super.recordTransactions(txs) + super.recordTransactions(notifyVault, txs) } } override fun jdbcSession(): Connection = database.createSession() diff --git a/node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt b/node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt index 192f4a0b55..a3989324a0 100644 --- a/node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt +++ b/node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt @@ -89,8 +89,8 @@ interface ServiceHubInternal : PluginServiceHub { val database: CordaPersistence val configuration: NodeConfiguration - override fun recordTransactions(txs: Iterable) { - require (txs.any()) { "No transactions passed in for recording" } + override fun recordTransactions(notifyVault: Boolean, txs: Iterable) { + require(txs.any()) { "No transactions passed in for recording" } val recordedTransactions = txs.filter { validatedTransactions.addTransaction(it) } val stateMachineRunId = FlowStateMachineImpl.currentStateMachine()?.id if (stateMachineRunId != null) { @@ -101,8 +101,10 @@ interface ServiceHubInternal : PluginServiceHub { log.warn("Transactions recorded from outside of a state machine") } - val toNotify = recordedTransactions.map { if (it.isNotaryChangeTransaction()) it.notaryChangeTx else it.tx } - vaultService.notifyAll(toNotify) + if (notifyVault) { + val toNotify = recordedTransactions.map { if (it.isNotaryChangeTransaction()) it.notaryChangeTx else it.tx } + vaultService.notifyAll(toNotify) + } } /** diff --git a/node/src/main/kotlin/net/corda/node/services/vault/HibernateVaultQueryImpl.kt b/node/src/main/kotlin/net/corda/node/services/vault/HibernateVaultQueryImpl.kt index a8a58bd2a8..55cd451b38 100644 --- a/node/src/main/kotlin/net/corda/node/services/vault/HibernateVaultQueryImpl.kt +++ b/node/src/main/kotlin/net/corda/node/services/vault/HibernateVaultQueryImpl.kt @@ -28,7 +28,6 @@ import java.lang.Exception import java.util.* import javax.persistence.Tuple - class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration, val vault: VaultService) : SingletonSerializeAsToken(), VaultQueryService { companion object { diff --git a/node/src/test/kotlin/net/corda/node/services/database/HibernateConfigurationTest.kt b/node/src/test/kotlin/net/corda/node/services/database/HibernateConfigurationTest.kt index c9abfefbe1..1f9c6de17b 100644 --- a/node/src/test/kotlin/net/corda/node/services/database/HibernateConfigurationTest.kt +++ b/node/src/test/kotlin/net/corda/node/services/database/HibernateConfigurationTest.kt @@ -75,7 +75,7 @@ class HibernateConfigurationTest : TestDependencyInjectionBase() { services = object : MockServices(BOB_KEY, BOC_KEY, DUMMY_NOTARY_KEY) { override val vaultService: VaultService = makeVaultService(dataSourceProps, hibernateConfig) - override fun recordTransactions(txs: Iterable) { + override fun recordTransactions(notifyVault: Boolean, txs: Iterable) { for (stx in txs) { validatedTransactions.addTransaction(stx) } diff --git a/node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt b/node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt index 62e1591b15..91587226b2 100644 --- a/node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt +++ b/node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt @@ -94,13 +94,14 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() { val originalVaultQuery = vaultQuery val services2 = object : MockServices() { override val vaultService: VaultService get() = originalVault - override fun recordTransactions(txs: Iterable) { + override fun recordTransactions(notifyVault: Boolean, txs: Iterable) { for (stx in txs) { validatedTransactions.addTransaction(stx) vaultService.notify(stx.tx) } } - override val vaultQueryService : VaultQueryService get() = originalVaultQuery + + override val vaultQueryService: VaultQueryService get() = originalVaultQuery } val w2 = services2.vaultQueryService.queryBy().states diff --git a/test-utils/src/main/kotlin/net/corda/testing/node/MockServices.kt b/test-utils/src/main/kotlin/net/corda/testing/node/MockServices.kt index c468025bda..bffbf0ea2a 100644 --- a/test-utils/src/main/kotlin/net/corda/testing/node/MockServices.kt +++ b/test-utils/src/main/kotlin/net/corda/testing/node/MockServices.kt @@ -60,7 +60,7 @@ open class MockServices(vararg val keys: KeyPair) : ServiceHub { val key: KeyPair get() = keys.first() - override fun recordTransactions(txs: Iterable) { + override fun recordTransactions(notifyVault: Boolean, txs: Iterable) { txs.forEach { stateMachineRecordedTransactionMapping.addMapping(StateMachineRunId.createRandom(), it.id) } @@ -229,7 +229,7 @@ fun makeTestDatabaseAndMockServices(customSchemas: Set = setOf(Com object : MockServices(*(keys.toTypedArray())) { override val vaultService: VaultService = makeVaultService(dataSourceProps, hibernateConfig) - override fun recordTransactions(txs: Iterable) { + override fun recordTransactions(notifyVault: Boolean, txs: Iterable) { for (stx in txs) { validatedTransactions.addTransaction(stx) }