mirror of
https://github.com/corda/corda.git
synced 2025-01-02 19:26:47 +00:00
Replace Vault.PageAndUpdates with DataFeed data class (#931)
* Replace kotlin Pair with DataFeed data class * remove unintended changes * Replace Vault.PageAndUpdates with DataFeed data class * Remove PageAndUpdates
This commit is contained in:
parent
e02c37c06d
commit
c3ca2744aa
@ -125,23 +125,23 @@ interface CordaRPCOps : RPCOps {
|
|||||||
fun <T : ContractState> vaultTrackBy(criteria: QueryCriteria,
|
fun <T : ContractState> vaultTrackBy(criteria: QueryCriteria,
|
||||||
paging: PageSpecification,
|
paging: PageSpecification,
|
||||||
sorting: Sort,
|
sorting: Sort,
|
||||||
contractType: Class<out T>): Vault.PageAndUpdates<T>
|
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update>
|
||||||
// DOCEND VaultTrackByAPI
|
// DOCEND VaultTrackByAPI
|
||||||
|
|
||||||
// Note: cannot apply @JvmOverloads to interfaces nor interface implementations
|
// Note: cannot apply @JvmOverloads to interfaces nor interface implementations
|
||||||
// Java Helpers
|
// Java Helpers
|
||||||
|
|
||||||
// DOCSTART VaultTrackAPIHelpers
|
// DOCSTART VaultTrackAPIHelpers
|
||||||
fun <T : ContractState> vaultTrack(contractType: Class<out T>): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> vaultTrack(contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return vaultTrackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), contractType)
|
return vaultTrackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> vaultTrackByCriteria(contractType: Class<out T>, criteria: QueryCriteria): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> vaultTrackByCriteria(contractType: Class<out T>, criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return vaultTrackBy(criteria, PageSpecification(), Sort(emptySet()), contractType)
|
return vaultTrackBy(criteria, PageSpecification(), Sort(emptySet()), contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> vaultTrackByWithPagingSpec(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> vaultTrackByWithPagingSpec(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return vaultTrackBy(criteria, paging, Sort(emptySet()), contractType)
|
return vaultTrackBy(criteria, paging, Sort(emptySet()), contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> vaultTrackByWithSorting(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> vaultTrackByWithSorting(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return vaultTrackBy(criteria, PageSpecification(), sorting, contractType)
|
return vaultTrackBy(criteria, PageSpecification(), sorting, contractType)
|
||||||
}
|
}
|
||||||
// DOCEND VaultTrackAPIHelpers
|
// DOCEND VaultTrackAPIHelpers
|
||||||
@ -302,7 +302,7 @@ inline fun <reified T : ContractState> CordaRPCOps.vaultQueryBy(criteria: QueryC
|
|||||||
|
|
||||||
inline fun <reified T : ContractState> CordaRPCOps.vaultTrackBy(criteria: QueryCriteria = QueryCriteria.VaultQueryCriteria(),
|
inline fun <reified T : ContractState> CordaRPCOps.vaultTrackBy(criteria: QueryCriteria = QueryCriteria.VaultQueryCriteria(),
|
||||||
paging: PageSpecification = PageSpecification(),
|
paging: PageSpecification = PageSpecification(),
|
||||||
sorting: Sort = Sort(emptySet())): Vault.PageAndUpdates<T> {
|
sorting: Sort = Sort(emptySet())): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return vaultTrackBy(criteria, paging, sorting, T::class.java)
|
return vaultTrackBy(criteria, paging, sorting, T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -402,4 +402,8 @@ data class DataFeed<out A, B>(val snapshot: A, val updates: Observable<B>) {
|
|||||||
val first: A get() = snapshot
|
val first: A get() = snapshot
|
||||||
@Deprecated("This function will be removed in a future milestone", ReplaceWith("updates"))
|
@Deprecated("This function will be removed in a future milestone", ReplaceWith("updates"))
|
||||||
val second: Observable<B> get() = updates
|
val second: Observable<B> get() = updates
|
||||||
|
@Deprecated("This function will be removed in a future milestone", ReplaceWith("snapshot"))
|
||||||
|
val current: A get() = snapshot
|
||||||
|
@Deprecated("This function will be removed in a future milestone", ReplaceWith("updates"))
|
||||||
|
val future: Observable<B> get() = updates
|
||||||
}
|
}
|
||||||
|
@ -141,9 +141,6 @@ class Vault<out T : ContractState>(val states: Iterable<StateAndRef<T>>) {
|
|||||||
val notaryKey: String,
|
val notaryKey: String,
|
||||||
val lockId: String?,
|
val lockId: String?,
|
||||||
val lockUpdateTime: Instant?)
|
val lockUpdateTime: Instant?)
|
||||||
|
|
||||||
@CordaSerializable
|
|
||||||
data class PageAndUpdates<out T : ContractState>(val current: Vault.Page<T>, val future: Observable<Vault.Update>)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -382,7 +379,7 @@ interface VaultQueryService {
|
|||||||
fun <T : ContractState> _trackBy(criteria: QueryCriteria,
|
fun <T : ContractState> _trackBy(criteria: QueryCriteria,
|
||||||
paging: PageSpecification,
|
paging: PageSpecification,
|
||||||
sorting: Sort,
|
sorting: Sort,
|
||||||
contractType: Class<out T>): Vault.PageAndUpdates<T>
|
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update>
|
||||||
// DOCEND VaultQueryAPI
|
// DOCEND VaultQueryAPI
|
||||||
|
|
||||||
// Note: cannot apply @JvmOverloads to interfaces nor interface implementations
|
// Note: cannot apply @JvmOverloads to interfaces nor interface implementations
|
||||||
@ -406,16 +403,16 @@ interface VaultQueryService {
|
|||||||
fun <T : ContractState> trackBy(contractType: Class<out T>): Vault.Page<T> {
|
fun <T : ContractState> trackBy(contractType: Class<out T>): Vault.Page<T> {
|
||||||
return _queryBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), contractType)
|
return _queryBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, PageSpecification(), Sort(emptySet()), contractType)
|
return _trackBy(criteria, PageSpecification(), Sort(emptySet()), contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, paging, Sort(emptySet()), contractType)
|
return _trackBy(criteria, paging, Sort(emptySet()), contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, PageSpecification(), sorting, contractType)
|
return _trackBy(criteria, PageSpecification(), sorting, contractType)
|
||||||
}
|
}
|
||||||
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): Vault.PageAndUpdates<T> {
|
fun <T : ContractState> trackBy(contractType: Class<out T>, criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, paging, sorting, contractType)
|
return _trackBy(criteria, paging, sorting, contractType)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -440,23 +437,23 @@ inline fun <reified T : ContractState> VaultQueryService.queryBy(criteria: Query
|
|||||||
return _queryBy(criteria, paging, sorting, T::class.java)
|
return _queryBy(criteria, paging, sorting, T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
inline fun <reified T : ContractState> VaultQueryService.trackBy(): Vault.PageAndUpdates<T> {
|
inline fun <reified T : ContractState> VaultQueryService.trackBy(): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), T::class.java)
|
return _trackBy(QueryCriteria.VaultQueryCriteria(), PageSpecification(), Sort(emptySet()), T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria): Vault.PageAndUpdates<T> {
|
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, PageSpecification(), Sort(emptySet()), T::class.java)
|
return _trackBy(criteria, PageSpecification(), Sort(emptySet()), T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification): Vault.PageAndUpdates<T> {
|
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, paging, Sort(emptySet()), T::class.java)
|
return _trackBy(criteria, paging, Sort(emptySet()), T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, sorting: Sort): Vault.PageAndUpdates<T> {
|
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, PageSpecification(), sorting, T::class.java)
|
return _trackBy(criteria, PageSpecification(), sorting, T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): Vault.PageAndUpdates<T> {
|
inline fun <reified T : ContractState> VaultQueryService.trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return _trackBy(criteria, paging, sorting, T::class.java)
|
return _trackBy(criteria, paging, sorting, T::class.java)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -68,9 +68,9 @@ class CordaRPCOpsImpl(
|
|||||||
override fun <T : ContractState> vaultTrackBy(criteria: QueryCriteria,
|
override fun <T : ContractState> vaultTrackBy(criteria: QueryCriteria,
|
||||||
paging: PageSpecification,
|
paging: PageSpecification,
|
||||||
sorting: Sort,
|
sorting: Sort,
|
||||||
contractType: Class<out T>): Vault.PageAndUpdates<T> {
|
contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return database.transaction {
|
return database.transaction {
|
||||||
services.vaultQueryService._trackBy<T>(criteria, paging, sorting, contractType)
|
services.vaultQueryService._trackBy(criteria, paging, sorting, contractType)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -7,6 +7,7 @@ import net.corda.core.contracts.StateAndRef
|
|||||||
import net.corda.core.contracts.StateRef
|
import net.corda.core.contracts.StateRef
|
||||||
import net.corda.core.contracts.TransactionState
|
import net.corda.core.contracts.TransactionState
|
||||||
import net.corda.core.crypto.SecureHash
|
import net.corda.core.crypto.SecureHash
|
||||||
|
import net.corda.core.messaging.DataFeed
|
||||||
import net.corda.core.node.services.Vault
|
import net.corda.core.node.services.Vault
|
||||||
import net.corda.core.node.services.VaultQueryException
|
import net.corda.core.node.services.VaultQueryException
|
||||||
import net.corda.core.node.services.VaultQueryService
|
import net.corda.core.node.services.VaultQueryService
|
||||||
@ -20,7 +21,6 @@ import net.corda.core.serialization.storageKryo
|
|||||||
import net.corda.core.utilities.loggerFor
|
import net.corda.core.utilities.loggerFor
|
||||||
import net.corda.node.services.database.HibernateConfiguration
|
import net.corda.node.services.database.HibernateConfiguration
|
||||||
import net.corda.node.services.vault.schemas.jpa.VaultSchemaV1
|
import net.corda.node.services.vault.schemas.jpa.VaultSchemaV1
|
||||||
import net.corda.node.utilities.wrapWithDatabaseTransaction
|
|
||||||
import org.jetbrains.exposed.sql.transactions.TransactionManager
|
import org.jetbrains.exposed.sql.transactions.TransactionManager
|
||||||
import rx.subjects.PublishSubject
|
import rx.subjects.PublishSubject
|
||||||
import java.lang.Exception
|
import java.lang.Exception
|
||||||
@ -99,15 +99,14 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private val mutex = ThreadBox ({ updatesPublisher })
|
private val mutex = ThreadBox({ updatesPublisher })
|
||||||
|
|
||||||
@Throws(VaultQueryException::class)
|
@Throws(VaultQueryException::class)
|
||||||
override fun <T : ContractState> _trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort, contractType: Class<out T>): Vault.PageAndUpdates<T> {
|
override fun <T : ContractState> _trackBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort, contractType: Class<out T>): DataFeed<Vault.Page<T>, Vault.Update> {
|
||||||
return mutex.locked {
|
return mutex.locked {
|
||||||
val snapshotResults = _queryBy<T>(criteria, paging, sorting, contractType)
|
val snapshotResults = _queryBy<T>(criteria, paging, sorting, contractType)
|
||||||
Vault.PageAndUpdates(snapshotResults,
|
val updates = updatesPublisher.bufferUntilSubscribed().filter { it.containsType(contractType, snapshotResults.stateTypes) }
|
||||||
updatesPublisher.bufferUntilSubscribed()
|
DataFeed(snapshotResults, updates)
|
||||||
.filter { it.containsType(contractType, snapshotResults.stateTypes) } )
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -115,7 +114,7 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
|
|||||||
* Maintain a list of contract state interfaces to concrete types stored in the vault
|
* Maintain a list of contract state interfaces to concrete types stored in the vault
|
||||||
* for usage in generic queries of type queryBy<LinearState> or queryBy<FungibleState<*>>
|
* for usage in generic queries of type queryBy<LinearState> or queryBy<FungibleState<*>>
|
||||||
*/
|
*/
|
||||||
fun resolveUniqueContractStateTypes(session: EntityManager) : Map<String, List<String>> {
|
fun resolveUniqueContractStateTypes(session: EntityManager): Map<String, List<String>> {
|
||||||
val criteria = criteriaBuilder.createQuery(String::class.java)
|
val criteria = criteriaBuilder.createQuery(String::class.java)
|
||||||
val vaultStates = criteria.from(VaultSchemaV1.VaultStates::class.java)
|
val vaultStates = criteria.from(VaultSchemaV1.VaultStates::class.java)
|
||||||
criteria.select(vaultStates.get("contractStateClassName")).distinct(true)
|
criteria.select(vaultStates.get("contractStateClassName")).distinct(true)
|
||||||
@ -135,7 +134,7 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
|
|||||||
return contractInterfaceToConcreteTypes
|
return contractInterfaceToConcreteTypes
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun <T: ContractState> deriveContractInterfaces(clazz: Class<T>): Set<Class<T>> {
|
private fun <T : ContractState> deriveContractInterfaces(clazz: Class<T>): Set<Class<T>> {
|
||||||
val myInterfaces: MutableSet<Class<T>> = mutableSetOf()
|
val myInterfaces: MutableSet<Class<T>> = mutableSetOf()
|
||||||
clazz.interfaces.forEach {
|
clazz.interfaces.forEach {
|
||||||
if (!it.equals(ContractState::class.java)) {
|
if (!it.equals(ContractState::class.java)) {
|
||||||
|
@ -7,6 +7,7 @@ import net.corda.contracts.asset.*;
|
|||||||
import net.corda.core.contracts.*;
|
import net.corda.core.contracts.*;
|
||||||
import net.corda.core.crypto.*;
|
import net.corda.core.crypto.*;
|
||||||
import net.corda.core.identity.*;
|
import net.corda.core.identity.*;
|
||||||
|
import net.corda.core.messaging.DataFeed;
|
||||||
import net.corda.core.node.services.*;
|
import net.corda.core.node.services.*;
|
||||||
import net.corda.core.node.services.vault.*;
|
import net.corda.core.node.services.vault.*;
|
||||||
import net.corda.core.node.services.vault.QueryCriteria.*;
|
import net.corda.core.node.services.vault.QueryCriteria.*;
|
||||||
@ -248,7 +249,7 @@ public class VaultQueryJavaTests {
|
|||||||
Set<Class<ContractState>> contractStateTypes = new HashSet(Collections.singletonList(Cash.State.class));
|
Set<Class<ContractState>> contractStateTypes = new HashSet(Collections.singletonList(Cash.State.class));
|
||||||
|
|
||||||
VaultQueryCriteria criteria = new VaultQueryCriteria(Vault.StateStatus.UNCONSUMED, contractStateTypes);
|
VaultQueryCriteria criteria = new VaultQueryCriteria(Vault.StateStatus.UNCONSUMED, contractStateTypes);
|
||||||
Vault.PageAndUpdates<ContractState> results = vaultQuerySvc.trackBy(ContractState.class, criteria);
|
DataFeed<Vault.Page<ContractState>, Vault.Update> results = vaultQuerySvc.trackBy(ContractState.class, criteria);
|
||||||
|
|
||||||
Vault.Page<ContractState> snapshot = results.getCurrent();
|
Vault.Page<ContractState> snapshot = results.getCurrent();
|
||||||
Observable<Vault.Update> updates = results.getFuture();
|
Observable<Vault.Update> updates = results.getFuture();
|
||||||
@ -289,7 +290,7 @@ public class VaultQueryJavaTests {
|
|||||||
PageSpecification pageSpec = new PageSpecification(0, getMAX_PAGE_SIZE());
|
PageSpecification pageSpec = new PageSpecification(0, getMAX_PAGE_SIZE());
|
||||||
Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC);
|
Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC);
|
||||||
Sort sorting = new Sort(ImmutableSet.of(sortByUid));
|
Sort sorting = new Sort(ImmutableSet.of(sortByUid));
|
||||||
Vault.PageAndUpdates<ContractState> results = vaultQuerySvc.trackBy(ContractState.class, compositeCriteria, pageSpec, sorting);
|
DataFeed<Vault.Page<ContractState>, Vault.Update> results = vaultQuerySvc.trackBy(ContractState.class, compositeCriteria, pageSpec, sorting);
|
||||||
|
|
||||||
Vault.Page<ContractState> snapshot = results.getCurrent();
|
Vault.Page<ContractState> snapshot = results.getCurrent();
|
||||||
Observable<Vault.Update> updates = results.getFuture();
|
Observable<Vault.Update> updates = results.getFuture();
|
||||||
|
Loading…
Reference in New Issue
Block a user