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

Add generics to Vault.Update

Browse Source
This commit is contained in:
Mike Hearn 2017-07-10 17:28:43 +02:00
parent dcda0b078e
commit a9a50ce8f3
17 changed files with 84 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
client/jfx/src/integration-test/kotlin/net/corda/client/jfx/NodeMonitorModelTest.kt
View File

@ -4,6 +4,7 @@ import net.corda.client.jfx.model.NodeMonitorModel
import net.corda.client.jfx.model.ProgressTrackingEvent
import net.corda.core.internal.bufferUntilSubscribed
import net.corda.core.contracts.Amount
import net.corda.core.contracts.ContractState
import net.corda.core.contracts.DOLLARS
import net.corda.core.contracts.USD
import net.corda.core.crypto.isFulfilledBy
@ -47,7 +48,7 @@ class NodeMonitorModelTest : DriverBasedTest() {
lateinit var stateMachineUpdatesBob: Observable<StateMachineUpdate>
lateinit var progressTracking: Observable<ProgressTrackingEvent>
lateinit var transactions: Observable<SignedTransaction>
lateinit var vaultUpdates: Observable<Vault.Update>
lateinit var vaultUpdates: Observable<Vault.Update<ContractState>>
lateinit var networkMapUpdates: Observable<NetworkMapCache.MapChange>
lateinit var newNode: (X500Name) -> NodeInfo

2
client/jfx/src/main/kotlin/net/corda/client/jfx/model/ContractStateModel.kt
View File

@ -19,7 +19,7 @@ data class Diff<out T : ContractState>(
* This model exposes the list of owned contract states.
*/
class ContractStateModel {
private val vaultUpdates: Observable<Vault.Update> by observable(NodeMonitorModel::vaultUpdates)
private val vaultUpdates: Observable<Vault.Update<ContractState>> by observable(NodeMonitorModel::vaultUpdates)
private val contractStatesDiff: Observable<Diff<ContractState>> = vaultUpdates.map {
Diff(it.produced, it.consumed)

7
client/jfx/src/main/kotlin/net/corda/client/jfx/model/NodeMonitorModel.kt
View File

@ -3,6 +3,7 @@ package net.corda.client.jfx.model
import javafx.beans.property.SimpleObjectProperty
import net.corda.client.rpc.CordaRPCClient
import net.corda.client.rpc.CordaRPCClientConfiguration
import net.corda.core.contracts.ContractState
import net.corda.core.flows.StateMachineRunId
import net.corda.core.messaging.CordaRPCOps
import net.corda.core.messaging.StateMachineInfo
@ -32,14 +33,14 @@ data class ProgressTrackingEvent(val stateMachineId: StateMachineRunId, val mess
class NodeMonitorModel {
private val stateMachineUpdatesSubject = PublishSubject.create<StateMachineUpdate>()
private val vaultUpdatesSubject = PublishSubject.create<Vault.Update>()
private val vaultUpdatesSubject = PublishSubject.create<Vault.Update<ContractState>>()
private val transactionsSubject = PublishSubject.create<SignedTransaction>()
private val stateMachineTransactionMappingSubject = PublishSubject.create<StateMachineTransactionMapping>()
private val progressTrackingSubject = PublishSubject.create<ProgressTrackingEvent>()
private val networkMapSubject = PublishSubject.create<MapChange>()
val stateMachineUpdates: Observable<StateMachineUpdate> = stateMachineUpdatesSubject
val vaultUpdates: Observable<Vault.Update> = vaultUpdatesSubject
val vaultUpdates: Observable<Vault.Update<ContractState>> = vaultUpdatesSubject
val transactions: Observable<SignedTransaction> = transactionsSubject
val stateMachineTransactionMapping: Observable<StateMachineTransactionMapping> = stateMachineTransactionMappingSubject
val progressTracking: Observable<ProgressTrackingEvent> = progressTrackingSubject
@ -85,7 +86,7 @@ class NodeMonitorModel {
// Vault updates
val (vault, vaultUpdates) = proxy.vaultAndUpdates()
val initialVaultUpdate = Vault.Update(setOf(), vault.toSet())
val initialVaultUpdate = Vault.Update<ContractState>(setOf(), vault.toSet())
vaultUpdates.startWith(initialVaultUpdate).subscribe(vaultUpdatesSubject)
// Transactions

14
core/src/main/kotlin/net/corda/core/messaging/CordaRPCOps.kt
View File

@ -136,23 +136,23 @@ interface CordaRPCOps : RPCOps {
fun <T : ContractState> vaultTrackBy(criteria: QueryCriteria,
paging: PageSpecification,
sorting: Sort,
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update>
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update<T>>
// DOCEND VaultTrackByAPI
// Note: cannot apply @JvmOverloads to interfaces nor interface implementations
// Java Helpers
// DOCSTART VaultTrackAPIHelpers
fun <T : ContractState> vaultTrack(contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> vaultTrack(contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return vaultTrackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), contractType)
}
fun <T : ContractState> vaultTrackByCriteria(contractType: Class<out T>, criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> vaultTrackByCriteria(contractType: Class<out T>, criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return vaultTrackBy(criteria, PageSpecification(), Sort(emptySet()), contractType)
}
fun <T : ContractState> vaultTrackByWithPagingSpec(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> vaultTrackByWithPagingSpec(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return vaultTrackBy(criteria, paging, Sort(emptySet()), contractType)
}
fun <T : ContractState> vaultTrackByWithSorting(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> vaultTrackByWithSorting(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return vaultTrackBy(criteria, PageSpecification(), sorting, contractType)
}
// DOCEND VaultTrackAPIHelpers
@ -163,7 +163,7 @@ interface CordaRPCOps : RPCOps {
@RPCReturnsObservables
// TODO: Remove this from the interface
@Deprecated("This function will be removed in a future milestone", ReplaceWith("vaultTrackBy(QueryCriteria())"))
fun vaultAndUpdates(): DataFeed<List<StateAndRef<ContractState>>, Vault.Update>
fun vaultAndUpdates(): DataFeed<List<StateAndRef<ContractState>>, Vault.Update<ContractState>>
/**
* Returns a data feed of all recorded transactions and an observable of future recorded ones.
@ -314,7 +314,7 @@ inline fun <reified T : ContractState> CordaRPCOps.vaultQueryBy(criteria: QueryC
inline fun <reified T : ContractState> CordaRPCOps.vaultTrackBy(criteria: QueryCriteria = QueryCriteria.VaultQueryCriteria(),
paging: PageSpecification = PageSpecification(),
sorting: Sort = Sort(emptySet())): DataFeed<Vault.Page<T>, Vault.Update> {
sorting: Sort = Sort(emptySet())): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return vaultTrackBy(criteria, paging, sorting, T::class.java)
}

24
core/src/main/kotlin/net/corda/core/node/services/VaultQueryService.kt
View File

@ -51,7 +51,7 @@ interface VaultQueryService {
fun <T : ContractState> _trackBy(criteria: QueryCriteria,
paging: PageSpecification,
sorting: Sort,
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update>
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update<T>>
// DOCEND VaultQueryAPI
// Note: cannot apply @JvmOverloads to interfaces nor interface implementations
@ -76,23 +76,23 @@ interface VaultQueryService {
return _queryBy(criteria, paging, sorting, contractType)
}
fun <T : ContractState> trackBy(contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> trackBy(contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), contractType)
}
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, PageSpecification(), Sort(emptySet()), contractType)
}
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, paging, Sort(emptySet()), contractType)
}
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, PageSpecification(), sorting, contractType)
}
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, paging, sorting, contractType)
}
}
@ -117,24 +117,24 @@ inline fun <reified T : ContractState> VaultQueryService.queryBy(criteria: Query
return _queryBy(criteria, paging, sorting, T::class.java)
}
inline fun <reified T : ContractState> VaultQueryService.trackBy(): DataFeed<Vault.Page<T>, Vault.Update> {
inline fun <reified T : ContractState> VaultQueryService.trackBy(): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), T::class.java)
}
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update> {
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, PageSpecification(), Sort(emptySet()), T::class.java)
}
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update> {
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, paging, Sort(emptySet()), T::class.java)
}
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, PageSpecification(), sorting, T::class.java)
}
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return _trackBy(criteria, paging, sorting, T::class.java)
}
class VaultQueryException(description: String) : FlowException("$description")
class VaultQueryException(description: String) : FlowException(description)

20
core/src/main/kotlin/net/corda/core/node/services/VaultService.kt
View File

@ -44,7 +44,7 @@ class Vault<out T : ContractState>(val states: Iterable<StateAndRef<T>>) {
* other transactions observed, then the changes are observed "net" of those.
*/
@CordaSerializable
data class Update(val consumed: Set<StateAndRef<ContractState>>, val produced: Set<StateAndRef<ContractState>>, val flowId: UUID? = null) {
data class Update<U : ContractState>(val consumed: Set<StateAndRef<U>>, val produced: Set<StateAndRef<U>>, val flowId: UUID? = null) {
/** Checks whether the update contains a state of the specified type. */
inline fun <reified T : ContractState> containsType() = consumed.any { it.state.data is T } || produced.any { it.state.data is T }
@ -63,8 +63,8 @@ class Vault<out T : ContractState>(val states: Iterable<StateAndRef<T>>) {
*
* i.e. the net effect in terms of state live-ness of receiving the combined update is the same as receiving this followed by rhs.
*/
operator fun plus(rhs: Update): Update {
val combined = Vault.Update(
operator fun plus(rhs: Update<U>): Update<U> {
val combined = Vault.Update<U>(
consumed + (rhs.consumed - produced),
// The ordering below matters to preserve ordering of consumed/produced Sets when they are insertion order dependent implementations.
produced.filter { it !in rhs.consumed }.toSet() + rhs.produced)
@ -84,7 +84,7 @@ class Vault<out T : ContractState>(val states: Iterable<StateAndRef<T>>) {
}
companion object {
val NoUpdate = Update(emptySet(), emptySet())
val NoUpdate = Update<ContractState>(emptySet(), emptySet())
}
@CordaSerializable
@ -143,18 +143,18 @@ interface VaultService {
* JVM or across to another [Thread] which is executing in a different database transaction) then the Vault may
* not incorporate the update due to racing with committing the current database transaction.
*/
val rawUpdates: Observable<Vault.Update>
val rawUpdates: Observable<Vault.Update<ContractState>>
/**
* Get a synchronous Observable of updates. When observations are pushed to the Observer, the Vault will already incorporate
* the update, and the database transaction associated with the update will have been committed and closed.
*/
val updates: Observable<Vault.Update>
val updates: Observable<Vault.Update<ContractState>>
/**
* Enable creation of observables of updates.
*/
val updatesPublisher: PublishSubject<Vault.Update>
val updatesPublisher: PublishSubject<Vault.Update<ContractState>>
/**
* Atomically get the current vault and a stream of updates. Note that the Observable buffers updates until the
@ -162,7 +162,7 @@ interface VaultService {
*/
// TODO: Remove this from the interface
@Deprecated("This function will be removed in a future milestone", ReplaceWith("trackBy(QueryCriteria())"))
fun track(): DataFeed<Vault<ContractState>, Vault.Update>
fun track(): DataFeed<Vault<ContractState>, Vault.Update<ContractState>>
/**
* Return unconsumed [ContractState]s for a given set of [StateRef]s
@ -185,7 +185,7 @@ interface VaultService {
/**
* Provide a [Future] for when a [StateRef] is consumed, which can be very useful in building tests.
*/
fun whenConsumed(ref: StateRef): ListenableFuture<Vault.Update> {
fun whenConsumed(ref: StateRef): ListenableFuture<Vault.Update<ContractState>> {
return updates.filter { it.consumed.any { it.ref == ref } }.toFuture()
}
@ -316,4 +316,4 @@ inline fun <reified T : LinearState> VaultService.linearHeadsOfType() =
class StatesNotAvailableException(override val message: String?, override val cause: Throwable? = null) : FlowException(message, cause) {
override fun toString() = "Soft locking error: $message"
}
}

24
core/src/test/kotlin/net/corda/core/node/VaultUpdateTests.kt
View File

@ -47,7 +47,7 @@ class VaultUpdateTests {
@Test
fun `something plus nothing is something`() {
val before = Vault.Update(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef2, stateAndRef3))
val before = Vault.Update<ContractState>(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef2, stateAndRef3))
val after = before + Vault.NoUpdate
assertEquals(before, after)
}
@ -55,32 +55,32 @@ class VaultUpdateTests {
@Test
fun `nothing plus something is something`() {
val before = Vault.NoUpdate
val after = before + Vault.Update(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef2, stateAndRef3))
val expected = Vault.Update(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef2, stateAndRef3))
val after = before + Vault.Update<ContractState>(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef2, stateAndRef3))
val expected = Vault.Update<ContractState>(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef2, stateAndRef3))
assertEquals(expected, after)
}
@Test
fun `something plus consume state 0 is something without state 0 output`() {
val before = Vault.Update(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1))
val after = before + Vault.Update(setOf(stateAndRef0), setOf())
val expected = Vault.Update(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef1))
val before = Vault.Update<ContractState>(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1))
val after = before + Vault.Update<ContractState>(setOf(stateAndRef0), setOf())
val expected = Vault.Update<ContractState>(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef1))
assertEquals(expected, after)
}
@Test
fun `something plus produce state 4 is something with additional state 4 output`() {
val before = Vault.Update(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1))
val after = before + Vault.Update(setOf(), setOf(stateAndRef4))
val expected = Vault.Update(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1, stateAndRef4))
val before = Vault.Update<ContractState>(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1))
val after = before + Vault.Update<ContractState>(setOf(), setOf(stateAndRef4))
val expected = Vault.Update<ContractState>(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1, stateAndRef4))
assertEquals(expected, after)
}
@Test
fun `something plus consume states 0 and 1, and produce state 4, is something without state 0 and 1 outputs and only state 4 output`() {
val before = Vault.Update(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1))
val after = before + Vault.Update(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef4))
val expected = Vault.Update(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef4))
val before = Vault.Update<ContractState>(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef0, stateAndRef1))
val after = before + Vault.Update<ContractState>(setOf(stateAndRef0, stateAndRef1), setOf(stateAndRef4))
val expected = Vault.Update<ContractState>(setOf(stateAndRef2, stateAndRef3), setOf(stateAndRef4))
assertEquals(expected, after)
}
}

9
docs/source/example-code/src/integration-test/kotlin/net/corda/docs/IntegrationTestingTutorial.kt
View File

@ -3,6 +3,7 @@ package net.corda.docs
import com.google.common.util.concurrent.Futures
import com.google.common.util.concurrent.ListenableFuture
import net.corda.contracts.asset.Cash
import net.corda.core.contracts.ContractState
import net.corda.core.contracts.DOLLARS
import net.corda.core.getOrThrow
import net.corda.core.messaging.startFlow
@ -82,8 +83,8 @@ class IntegrationTestingTutorial {
parallel(
(1..10).map { i ->
expect(
match = { update: Vault.Update ->
(update.produced.first().state.data as Cash.State).amount.quantity == i * 100L
match = { update: Vault.Update<Cash.State> ->
update.produced.first().state.data.amount.quantity == i * 100L
}
) { update ->
println("Bob vault update of $update")
@ -101,9 +102,9 @@ class IntegrationTestingTutorial {
aliceVaultUpdates.expectEvents {
sequence(
(1..10).map { i ->
expect { update: Vault.Update ->
expect { update: Vault.Update<Cash.State> ->
println("Alice got vault update of $update")
assertEquals((update.produced.first().state.data as Cash.State).amount.quantity, i * 100L)
assertEquals(update.produced.first().state.data.amount.quantity, i * 100L)
}
}
)

4
node/src/main/kotlin/net/corda/node/internal/CordaRPCOpsImpl.kt
View File

@ -45,7 +45,7 @@ class CordaRPCOpsImpl(
}
}
override fun vaultAndUpdates(): DataFeed<List<StateAndRef<ContractState>>, Vault.Update> {
override fun vaultAndUpdates(): DataFeed<List<StateAndRef<ContractState>>, Vault.Update<ContractState>> {
return database.transaction {
val (vault, updates) = services.vaultService.track()
DataFeed(vault.states.toList(), updates)
@ -65,7 +65,7 @@ class CordaRPCOpsImpl(
override fun <T : ContractState> vaultTrackBy(criteria: QueryCriteria,
paging: PageSpecification,
sorting: Sort,
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return database.transaction {
services.vaultQueryService._trackBy(criteria, paging, sorting, contractType)
}

2
node/src/main/kotlin/net/corda/node/services/schema/HibernateObserver.kt
View File

@ -17,7 +17,7 @@ import rx.Observable
* A vault observer that extracts Object Relational Mappings for contract states that support it, and persists them with Hibernate.
*/
// TODO: Manage version evolution of the schemas via additional tooling.
class HibernateObserver(vaultUpdates: Observable<Vault.Update>, val config: HibernateConfiguration) {
class HibernateObserver(vaultUpdates: Observable<Vault.Update<ContractState>>, val config: HibernateConfiguration) {
companion object {
val logger = loggerFor<HibernateObserver>()

11
node/src/main/kotlin/net/corda/node/services/vault/HibernateVaultQueryImpl.kt
View File

@ -21,6 +21,7 @@ import net.corda.core.utilities.loggerFor
import net.corda.node.services.database.HibernateConfiguration
import org.jetbrains.exposed.sql.transactions.TransactionManager
import rx.subjects.PublishSubject
import rx.Observable
import java.lang.Exception
import java.util.*
import javax.persistence.EntityManager
@ -28,7 +29,7 @@ import javax.persistence.Tuple
class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
val updatesPublisher: PublishSubject<Vault.Update>) : SingletonSerializeAsToken(), VaultQueryService {
val updatesPublisher: PublishSubject<Vault.Update<ContractState>>) : SingletonSerializeAsToken(), VaultQueryService {
companion object {
val log = loggerFor<HibernateVaultQueryImpl>()
}
@ -107,7 +108,6 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
}
return Vault.Page(states = statesAndRefs, statesMetadata = statesMeta, stateTypes = criteriaParser.stateTypes, totalStatesAvailable = totalStates, otherResults = otherResults)
} catch (e: Exception) {
log.error(e.message)
throw e.cause ?: e
@ -118,10 +118,11 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
private val mutex = ThreadBox({ updatesPublisher })
@Throws(VaultQueryException::class)
override fun <T : ContractState> _trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort, contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
override fun <T : ContractState> _trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort, contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update<T>> {
return mutex.locked {
val snapshotResults = _queryBy<T>(criteria, paging, sorting, contractType)
val updates = updatesPublisher.bufferUntilSubscribed().filter { it.containsType(contractType, snapshotResults.stateTypes) }
val snapshotResults = _queryBy(criteria, paging, sorting, contractType)
@Suppress("UNCHECKED_CAST")
val updates = updatesPublisher.bufferUntilSubscribed().filter { it.containsType(contractType, snapshotResults.stateTypes) } as Observable<Vault.Update<T>>
DataFeed(snapshotResults, updates)
}
}

18
node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt
View File

@ -74,16 +74,16 @@ class NodeVaultService(private val services: ServiceHub, dataSourceProperties: P
val session = configuration.sessionForModel(Models.VAULT)
private class InnerState {
val _updatesPublisher = PublishSubject.create<Vault.Update>()!!
val _rawUpdatesPublisher = PublishSubject.create<Vault.Update>()!!
val _updatesPublisher = PublishSubject.create<Vault.Update<ContractState>>()!!
val _rawUpdatesPublisher = PublishSubject.create<Vault.Update<ContractState>>()!!
val _updatesInDbTx = _updatesPublisher.wrapWithDatabaseTransaction().asObservable()!!
// For use during publishing only.
val updatesPublisher: rx.Observer<Vault.Update> get() = _updatesPublisher.bufferUntilDatabaseCommit().tee(_rawUpdatesPublisher)
val updatesPublisher: rx.Observer<Vault.Update<ContractState>> get() = _updatesPublisher.bufferUntilDatabaseCommit().tee(_rawUpdatesPublisher)
}
private val mutex = ThreadBox(InnerState())
private fun recordUpdate(update: Vault.Update): Vault.Update {
private fun recordUpdate(update: Vault.Update<ContractState>): Vault.Update<ContractState> {
if (update != Vault.NoUpdate) {
val producedStateRefs = update.produced.map { it.ref }
val producedStateRefsMap = update.produced.associateBy { it.ref }
@ -126,16 +126,16 @@ class NodeVaultService(private val services: ServiceHub, dataSourceProperties: P
return update
}
override val rawUpdates: Observable<Vault.Update>
override val rawUpdates: Observable<Vault.Update<ContractState>>
get() = mutex.locked { _rawUpdatesPublisher }
override val updates: Observable<Vault.Update>
override val updates: Observable<Vault.Update<ContractState>>
get() = mutex.locked { _updatesInDbTx }
override val updatesPublisher: PublishSubject<Vault.Update>
override val updatesPublisher: PublishSubject<Vault.Update<ContractState>>
get() = mutex.locked { _updatesPublisher }
override fun track(): DataFeed<Vault<ContractState>, Vault.Update> {
override fun track(): DataFeed<Vault<ContractState>, Vault.Update<ContractState>> {
return mutex.locked {
DataFeed(Vault(unconsumedStates<ContractState>()), _updatesPublisher.bufferUntilSubscribed().wrapWithDatabaseTransaction())
}
@ -421,7 +421,7 @@ class NodeVaultService(private val services: ServiceHub, dataSourceProperties: P
= txState.copy(data = txState.data.copy(amount = amount, owner = owner))
@VisibleForTesting
internal fun makeUpdate(tx: WireTransaction, ourKeys: Set<PublicKey>): Vault.Update {
internal fun makeUpdate(tx: WireTransaction, ourKeys: Set<PublicKey>): Vault.Update<ContractState> {
val ourNewStates = tx.filterOutRefs<ContractState> { isRelevant(it, ourKeys) }
// Retrieve all unconsumed states for this transaction's inputs

8
node/src/test/java/net/corda/node/services/vault/VaultQueryJavaTests.java
View File

@ -288,10 +288,10 @@ public class VaultQueryJavaTests extends TestDependencyInjectionBase {
Set<Class<ContractState>> contractStateTypes = new HashSet(Collections.singletonList(Cash.State.class));
VaultQueryCriteria criteria = new VaultQueryCriteria(Vault.StateStatus.UNCONSUMED, contractStateTypes);
DataFeed<Vault.Page<ContractState>, Vault.Update> results = vaultQuerySvc.trackBy(ContractState.class, criteria);
DataFeed<Vault.Page<ContractState>, Vault.Update<ContractState>> results = vaultQuerySvc.trackBy(ContractState.class, criteria);
Vault.Page<ContractState> snapshot = results.getSnapshot();
Observable<Vault.Update> updates = results.getUpdates();
Observable<Vault.Update<ContractState>> updates = results.getUpdates();
// DOCEND VaultJavaQueryExample4
assertThat(snapshot.getStates()).hasSize(3);
@ -325,10 +325,10 @@ public class VaultQueryJavaTests extends TestDependencyInjectionBase {
PageSpecification pageSpec = new PageSpecification(DEFAULT_PAGE_NUM, MAX_PAGE_SIZE);
Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC);
Sort sorting = new Sort(ImmutableSet.of(sortByUid));
DataFeed<Vault.Page<ContractState>, Vault.Update> results = vaultQuerySvc.trackBy(ContractState.class, compositeCriteria, pageSpec, sorting);
DataFeed<Vault.Page<ContractState>, Vault.Update<ContractState>> results = vaultQuerySvc.trackBy(ContractState.class, compositeCriteria, pageSpec, sorting);
Vault.Page<ContractState> snapshot = results.getSnapshot();
Observable<Vault.Update> updates = results.getUpdates();
Observable<Vault.Update<ContractState>> updates = results.getUpdates();
// DOCEND VaultJavaQueryExample5
assertThat(snapshot.getStates()).hasSize(13);

4
node/src/test/kotlin/net/corda/node/CordaRPCOpsImplTest.kt
View File

@ -53,8 +53,8 @@ class CordaRPCOpsImplTest {
lateinit var rpc: CordaRPCOps
lateinit var stateMachineUpdates: Observable<StateMachineUpdate>
lateinit var transactions: Observable<SignedTransaction>
lateinit var vaultUpdates: Observable<Vault.Update> // TODO: deprecated
lateinit var vaultTrackCash: Observable<Vault.Update>
lateinit var vaultUpdates: Observable<Vault.Update<ContractState>> // TODO: deprecated
lateinit var vaultTrackCash: Observable<Vault.Update<Cash.State>>
@Before
fun setup() {

2
node/src/test/kotlin/net/corda/node/services/schema/HibernateObserverTests.kt
View File

@ -86,7 +86,7 @@ class HibernateObserverTests {
@Test
fun testChildObjectsArePersisted() {
val testSchema = object : MappedSchema(SchemaFamily::class.java, 1, setOf(Parent::class.java, Child::class.java)) {}
val rawUpdatesPublisher = PublishSubject.create<Vault.Update>()
val rawUpdatesPublisher = PublishSubject.create<Vault.Update<ContractState>>()
val schemaService = object : SchemaService {
override val schemaOptions: Map<MappedSchema, SchemaService.SchemaOptions> = emptyMap()

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

@ -449,7 +449,7 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
@Test
fun `make update`() {
val service = (services.vaultService as NodeVaultService)
val vaultSubscriber = TestSubscriber<Vault.Update>().apply {
val vaultSubscriber = TestSubscriber<Vault.Update<*>>().apply {
service.updates.subscribe(this)
}

5
test-utils/src/main/kotlin/net/corda/testing/Expect.kt
View File

@ -78,6 +78,11 @@ fun <E> sequence(expectations: List<ExpectCompose<E>>): ExpectCompose<E> = Expec
*/
fun <E> parallel(vararg expectations: ExpectCompose<E>): ExpectCompose<E> = ExpectCompose.Parallel(listOf(*expectations))
/**
* Tests that events arrive in unspecified order.
*
* @param expectations The pieces of DSL all of which should run but in an unspecified order depending on what sequence events arrive.
*/
fun <E> parallel(expectations: List<ExpectCompose<E>>): ExpectCompose<E> = ExpectCompose.Parallel(expectations)
/**