Skip to content

Commit ff99df7

Browse files
roastarioroastario
committed
refactor some code out of VaultWatcherService.kt
1 parent db195dd commit ff99df7

File tree

7 files changed

+138
-114
lines changed

7 files changed

+138
-114
lines changed

‎modules/selection/src/main/kotlin/com.r3.corda.lib.tokens.selection/memory/config/InMemorySelectionConfig.kt

+10-9
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package com.r3.corda.lib.tokens.selection.memory.config
33
import co.paralleluniverse.fibers.Suspendable
44
import com.r3.corda.lib.tokens.selection.api.StateSelectionConfig
55
import com.r3.corda.lib.tokens.selection.memory.selector.LocalTokenSelector
6+
import com.r3.corda.lib.tokens.selection.memory.services.IndexingType
67
import com.r3.corda.lib.tokens.selection.memory.services.VaultWatcherService
78
import net.corda.core.cordapp.CordappConfig
89
import net.corda.core.cordapp.CordappConfigException
@@ -13,12 +14,12 @@ const val CACHE_SIZE_DEFAULT = 1024
1314
const val PAGE_SIZE_DEFAULT = 1024
1415

1516
data class InMemorySelectionConfig @JvmOverloads constructor(
16-
val enabled: Boolean,
17-
val indexingStrategies: List<VaultWatcherService.IndexingType>,
18-
val cacheSize: Int = CACHE_SIZE_DEFAULT,
19-
val pageSize: Int = 1000,
20-
val sleep: Int = 0,
21-
val loadingThreads: Int = 4
17+
val enabled: Boolean,
18+
val indexingStrategies: List<IndexingType>,
19+
val cacheSize: Int = CACHE_SIZE_DEFAULT,
20+
val pageSize: Int = 1000,
21+
val sleep: Int = 0,
22+
val loadingThreads: Int = 4
2223
) : StateSelectionConfig {
2324
companion object {
2425
private val logger = LoggerFactory.getLogger("inMemoryConfigSelectionLogger")
@@ -37,13 +38,13 @@ data class InMemorySelectionConfig @JvmOverloads constructor(
3738
val loadingSleep: Int = config.getIntOrNull("stateSelection.inMemory.loadingSleepSeconds")?: 0
3839
val loadingThreads: Int = config.getIntOrNull("stateSelection.inMemory.loadingThreads")?: 4
3940
val indexingType = try {
40-
(config.get("stateSelection.inMemory.indexingStrategies") as List<Any>).map { VaultWatcherService.IndexingType.valueOf(it.toString()) }
41+
(config.get("stateSelection.inMemory.indexingStrategies") as List<Any>).map { IndexingType.valueOf(it.toString()) }
4142
} catch (e: CordappConfigException) {
4243
logger.warn("No indexing method specified. Indexes will be created at run-time for each invocation of selectTokens")
43-
emptyList<VaultWatcherService.IndexingType>()
44+
emptyList<IndexingType>()
4445
} catch (e: ClassCastException) {
4546
logger.warn("No indexing method specified. Indexes will be created at run-time for each invocation of selectTokens")
46-
emptyList<VaultWatcherService.IndexingType>()
47+
emptyList<IndexingType>()
4748
}
4849
logger.info("Found in memory token selection configuration with values indexing strategy: $indexingType, cacheSize: $cacheSize")
4950
return InMemorySelectionConfig(enabled, indexingType, cacheSize, pageSize, loadingSleep, loadingThreads)

‎modules/selection/src/main/kotlin/com.r3.corda.lib.tokens.selection/memory/services/IndexingType.kt

+25
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
package com.r3.corda.lib.tokens.selection.memory.services
2+
3+
import com.r3.corda.lib.tokens.selection.memory.internal.Holder
4+
5+
enum class IndexingType(val ownerType: Class<out Holder>) {
6+
7+
EXTERNAL_ID(Holder.MappedIdentity::class.java),
8+
PUBLIC_KEY(Holder.KeyIdentity::class.java);
9+
10+
companion object {
11+
fun fromHolder(holder: Class<out Holder>): IndexingType {
12+
return when (holder) {
13+
Holder.MappedIdentity::class.java -> {
14+
EXTERNAL_ID
15+
}
16+
17+
Holder.KeyIdentity::class.java -> {
18+
PUBLIC_KEY
19+
}
20+
else -> throw IllegalArgumentException("Unknown Holder type: $holder")
21+
}
22+
}
23+
}
24+
25+
}

‎modules/selection/src/main/kotlin/com.r3.corda.lib.tokens.selection/memory/services/ServiceHubAsyncLoader.kt

+75
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
package com.r3.corda.lib.tokens.selection.memory.services
2+
3+
import com.r3.corda.lib.tokens.contracts.states.FungibleToken
4+
import com.r3.corda.lib.tokens.selection.memory.config.InMemorySelectionConfig
5+
import com.r3.corda.lib.tokens.selection.sortByTimeStampAscending
6+
import io.github.classgraph.ClassGraph
7+
import net.corda.core.node.AppServiceHub
8+
import net.corda.core.node.services.Vault
9+
import net.corda.core.node.services.queryBy
10+
import net.corda.core.node.services.vault.DEFAULT_PAGE_NUM
11+
import net.corda.core.node.services.vault.PageSpecification
12+
import net.corda.core.node.services.vault.QueryCriteria
13+
import net.corda.core.utilities.contextLogger
14+
import java.util.concurrent.CompletableFuture
15+
import java.util.concurrent.atomic.AtomicBoolean
16+
import java.util.concurrent.atomic.AtomicInteger
17+
18+
class ServiceHubAsyncLoader(private val appServiceHub: AppServiceHub,
19+
private val configOptions: InMemorySelectionConfig) : ((Vault.Update<FungibleToken>) -> Unit) -> Unit {
20+
21+
22+
override fun invoke(
23+
onVaultUpdate: (Vault.Update<FungibleToken>) -> Unit
24+
) {
25+
LOG.info("Starting async token loading from vault")
26+
27+
val classGraph = ClassGraph()
28+
classGraph.enableClassInfo()
29+
30+
val scanResultFuture = CompletableFuture.supplyAsync {
31+
classGraph.scan()
32+
}
33+
34+
scanResultFuture.thenApplyAsync { scanResult ->
35+
val subclasses: Set<Class<out FungibleToken>> = scanResult.getSubclasses(FungibleToken::class.java.canonicalName)
36+
.map { it.name }
37+
.map { Class.forName(it) as Class<out FungibleToken> }.toSet()
38+
39+
val enrichedClasses = (subclasses - setOf(FungibleToken::class.java))
40+
LOG.info("Enriching token query with types: $enrichedClasses")
41+
42+
val shouldLoop = AtomicBoolean(true)
43+
val pageNumber = AtomicInteger(DEFAULT_PAGE_NUM - 1)
44+
val loadingFutures: List<CompletableFuture<Void>> = 0.until(configOptions.loadingThreads).map {
45+
CompletableFuture.runAsync {
46+
try {
47+
while (shouldLoop.get()) {
48+
val newlyLoadedStates = appServiceHub.vaultService.queryBy<FungibleToken>(
49+
paging = PageSpecification(pageNumber = pageNumber.addAndGet(1), pageSize = configOptions.pageSize),
50+
criteria = QueryCriteria.VaultQueryCriteria(contractStateTypes = subclasses),
51+
sorting = sortByTimeStampAscending()
52+
).states.toSet()
53+
onVaultUpdate(Vault.Update(emptySet(), newlyLoadedStates))
54+
LOG.info("publishing ${newlyLoadedStates.size} to async state loading callback")
55+
shouldLoop.compareAndSet(true, newlyLoadedStates.isNotEmpty())
56+
LOG.debug("shouldLoop=${shouldLoop}")
57+
if (configOptions.sleep > 0) {
58+
Thread.sleep(configOptions.sleep.toLong() * 1000)
59+
}
60+
}
61+
} catch (t: Throwable) {
62+
LOG.error("Token Loading Failed due to: ", t)
63+
}
64+
}
65+
}
66+
CompletableFuture.allOf(*loadingFutures.toTypedArray()).thenRunAsync {
67+
LOG.info("finished token loading")
68+
}
69+
}
70+
}
71+
72+
companion object {
73+
val LOG = contextLogger()
74+
}
75+
}

‎modules/selection/src/main/kotlin/com.r3.corda.lib.tokens.selection/memory/services/VaultWatcherService.kt

+18-96
Original file line numberDiff line numberDiff line change
@@ -10,16 +10,12 @@ import com.r3.corda.lib.tokens.selection.memory.config.InMemorySelectionConfig
1010
import com.r3.corda.lib.tokens.selection.memory.internal.Holder
1111
import com.r3.corda.lib.tokens.selection.memory.internal.lookupExternalIdFromKey
1212
import com.r3.corda.lib.tokens.selection.sortByStateRefAscending
13-
import com.r3.corda.lib.tokens.selection.sortByTimeStampAscending
14-
import io.github.classgraph.ClassGraph
15-
import io.github.classgraph.ScanResult
1613
import net.corda.core.contracts.Amount
1714
import net.corda.core.contracts.StateAndRef
1815
import net.corda.core.internal.uncheckedCast
1916
import net.corda.core.node.AppServiceHub
2017
import net.corda.core.node.services.CordaService
2118
import net.corda.core.node.services.Vault
22-
import net.corda.core.node.services.queryBy
2319
import net.corda.core.node.services.vault.DEFAULT_PAGE_NUM
2420
import net.corda.core.node.services.vault.PageSpecification
2521
import net.corda.core.node.services.vault.QueryCriteria
@@ -28,12 +24,8 @@ import net.corda.core.utilities.contextLogger
2824
import rx.Observable
2925
import java.time.Duration
3026
import java.util.concurrent.*
31-
import java.util.concurrent.atomic.AtomicBoolean
32-
import java.util.concurrent.atomic.AtomicInteger
3327
import java.util.concurrent.atomic.AtomicReference
3428
import java.util.concurrent.locks.ReentrantReadWriteLock
35-
import java.util.function.Function
36-
import java.util.function.Supplier
3729
import kotlin.concurrent.read
3830
import kotlin.concurrent.write
3931

@@ -55,41 +47,28 @@ class VaultWatcherService(
5547
private val indexViewCreationLock: ReentrantReadWriteLock = ReentrantReadWriteLock()
5648
private val UPDATER = Executors.newSingleThreadScheduledExecutor()
5749

58-
enum class IndexingType(val ownerType: Class<out Holder>) {
59-
60-
EXTERNAL_ID(Holder.MappedIdentity::class.java),
61-
PUBLIC_KEY(Holder.KeyIdentity::class.java);
62-
63-
companion object {
64-
fun fromHolder(holder: Class<out Holder>): IndexingType {
65-
return when (holder) {
66-
Holder.MappedIdentity::class.java -> {
67-
EXTERNAL_ID
68-
}
69-
70-
Holder.KeyIdentity::class.java -> {
71-
PUBLIC_KEY
72-
}
73-
else -> throw IllegalArgumentException("Unknown Holder type: $holder")
74-
}
75-
}
76-
}
77-
78-
}
79-
8050
constructor(appServiceHub: AppServiceHub) : this(
8151
getObservableFromAppServiceHub(appServiceHub),
8252
InMemorySelectionConfig.parse(appServiceHub.getAppContext().config)
8353
)
8454

55+
init {
56+
addTokensToCache(tokenObserver.initialValues)
57+
tokenObserver.source.doOnError {
58+
LOG.error("received error from observable", it)
59+
}
60+
tokenObserver.source.subscribe(::onVaultUpdate)
61+
tokenObserver.startLoading(::onVaultUpdate)
62+
}
63+
8564
companion object {
8665
val LOG = contextLogger()
8766

8867
private fun getObservableFromAppServiceHub(appServiceHub: AppServiceHub): TokenObserver {
89-
val config = appServiceHub.cordappProvider.getAppContext().config
90-
val configOptions: InMemorySelectionConfig = InMemorySelectionConfig.parse(config)
68+
val rawConfig = appServiceHub.cordappProvider.getAppContext().config
69+
val parsedConfig: InMemorySelectionConfig = InMemorySelectionConfig.parse(rawConfig)
9170

92-
if (!configOptions.enabled) {
71+
if (!parsedConfig.enabled) {
9372
LOG.info("Disabling inMemory token selection - refer to documentation on how to enable")
9473
return TokenObserver(emptyList(), Observable.empty(), { _, _ ->
9574
Holder.UnmappedIdentity()
@@ -105,76 +84,19 @@ class VaultWatcherService(
10584
}
10685
}
10786
}
108-
10987
val (_, vaultObservable) = appServiceHub.vaultService.trackBy(
11088
contractStateType = FungibleToken::class.java,
11189
paging = PageSpecification(pageNumber = DEFAULT_PAGE_NUM, pageSize = 1),
11290
criteria = QueryCriteria.VaultQueryCriteria(status = Vault.StateStatus.ALL),
11391
sorting = sortByStateRefAscending()
11492
)
11593

116-
val pageSize = configOptions.pageSize
117-
val asyncLoader = object : ((Vault.Update<FungibleToken>) -> Unit) -> Unit {
118-
override fun invoke(callback: (Vault.Update<FungibleToken>) -> Unit) {
119-
LOG.info("Starting async token loading from vault")
120-
121-
val classGraph = ClassGraph()
122-
classGraph.enableClassInfo()
123-
124-
val scanResultFuture = CompletableFuture.supplyAsync {
125-
classGraph.scan()
126-
}
127-
128-
scanResultFuture.thenApplyAsync { scanResult ->
129-
val subclasses: Set<Class<out FungibleToken>> = scanResult.getSubclasses(FungibleToken::class.java.canonicalName)
130-
.map { it.name }
131-
.map { Class.forName(it) as Class<out FungibleToken> }.toSet()
132-
133-
val enrichedClasses = (subclasses - setOf(FungibleToken::class.java))
134-
LOG.info("Enriching token query with types: $enrichedClasses")
135-
136-
val shouldLoop = AtomicBoolean(true)
137-
val pageNumber = AtomicInteger(DEFAULT_PAGE_NUM - 1)
138-
val loadingFutures: List<CompletableFuture<Void>> = 0.until(configOptions.loadingThreads).map {
139-
CompletableFuture.runAsync {
140-
try {
141-
while (shouldLoop.get()) {
142-
val newlyLoadedStates = appServiceHub.vaultService.queryBy<FungibleToken>(
143-
paging = PageSpecification(pageNumber = pageNumber.addAndGet(1), pageSize = pageSize),
144-
criteria = QueryCriteria.VaultQueryCriteria(contractStateTypes = subclasses),
145-
sorting = sortByTimeStampAscending()
146-
).states.toSet()
147-
callback(Vault.Update(emptySet(), newlyLoadedStates))
148-
LOG.info("publishing ${newlyLoadedStates.size} to async state loading callback")
149-
shouldLoop.compareAndSet(newlyLoadedStates.isNotEmpty(), true)
150-
LOG.debug("shouldLoop=${shouldLoop}")
151-
if (configOptions.sleep > 0) {
152-
Thread.sleep(configOptions.sleep.toLong() * 1000)
153-
}
154-
155-
}
156-
LOG.info("finished token loading")
157-
} catch (t: Throwable) {
158-
LOG.error("Token Loading Failed due to: ", t)
159-
}
160-
}
161-
}
162-
}
163-
}
164-
}
165-
94+
val asyncLoader = ServiceHubAsyncLoader(appServiceHub, parsedConfig)
16695
return TokenObserver(emptyList(), uncheckedCast(vaultObservable), ownerProvider, asyncLoader)
16796
}
16897
}
16998

170-
init {
171-
addTokensToCache(tokenObserver.initialValues)
172-
tokenObserver.source.doOnError {
173-
LOG.error("received error from observable", it)
174-
}
175-
tokenObserver.startLoading(::onVaultUpdate)
176-
tokenObserver.source.subscribe(::onVaultUpdate)
177-
}
99+
178100

179101
private fun processToken(token: StateAndRef<FungibleToken>, indexingType: IndexingType): TokenIndex {
180102
val owner = tokenObserver.ownerProvider(token, indexingType)
@@ -375,10 +297,10 @@ class VaultWatcherService(
375297
}
376298

377299
class TokenObserver(
378-
val initialValues: List<StateAndRef<FungibleToken>>,
379-
val source: Observable<Vault.Update<FungibleToken>>,
380-
val ownerProvider: ((StateAndRef<FungibleToken>, VaultWatcherService.IndexingType) -> Holder),
381-
inline val asyncLoader: ((Vault.Update<FungibleToken>) -> Unit) -> Unit = { _ -> }
300+
val initialValues: List<StateAndRef<FungibleToken>>,
301+
val source: Observable<Vault.Update<FungibleToken>>,
302+
val ownerProvider: ((StateAndRef<FungibleToken>, IndexingType) -> Holder),
303+
inline val asyncLoader: ((Vault.Update<FungibleToken>) -> Unit) -> Unit = { _ -> }
382304
) {
383305

384306
fun startLoading(loadingCallBack: (Vault.Update<FungibleToken>) -> Unit) {

‎workflows-integration-test/src/main/kotlin/com/r3/corda/lib/tokens/integration/workflows/TestFlows.kt

-1
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@ import net.corda.core.utilities.unwrap
3535
import java.time.Duration
3636
import java.time.temporal.ChronoUnit
3737
import java.util.*
38-
import javax.swing.plaf.nimbus.State
3938

4039
// This is very simple test flow for DvP.
4140
@CordaSerializable

‎workflows/src/test/kotlin/com/r3/corda/lib/tokens/workflows/ConfigSelectionTest.kt

+5-4
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import com.r3.corda.lib.tokens.selection.database.selector.DatabaseTokenSelectio
88
import com.r3.corda.lib.tokens.selection.memory.config.CACHE_SIZE_DEFAULT
99
import com.r3.corda.lib.tokens.selection.memory.config.InMemorySelectionConfig
1010
import com.r3.corda.lib.tokens.selection.memory.selector.LocalTokenSelector
11+
import com.r3.corda.lib.tokens.selection.memory.services.IndexingType
1112
import com.r3.corda.lib.tokens.selection.memory.services.VaultWatcherService
1213
import com.typesafe.config.ConfigFactory
1314
import net.corda.core.identity.CordaX500Name
@@ -58,14 +59,14 @@ class ConfigSelectionTest {
5859
val config = ConfigFactory.parseString("stateSelection {\n" +
5960
"inMemory {\n" +
6061
"cacheSize: 9000\n" +
61-
"indexingStrategies: [${VaultWatcherService.IndexingType.PUBLIC_KEY}]\n" +
62+
"indexingStrategies: [${IndexingType.PUBLIC_KEY}]\n" +
6263
"}\n" +
6364
"}")
6465
val cordappConfig = TypesafeCordappConfig(config)
6566

6667
val inMemoryConfig = InMemorySelectionConfig.parse(cordappConfig)
6768
assertThat(inMemoryConfig.cacheSize).isEqualTo(9000)
68-
assertThat(inMemoryConfig.indexingStrategies).isEqualTo(listOf(VaultWatcherService.IndexingType.PUBLIC_KEY))
69+
assertThat(inMemoryConfig.indexingStrategies).isEqualTo(listOf(IndexingType.PUBLIC_KEY))
6970

7071
val selection = ConfigSelection.getPreferredSelection(services, cordappConfig)
7172
assertThat(selection).isInstanceOf(LocalTokenSelector::class.java)
@@ -76,13 +77,13 @@ class ConfigSelectionTest {
7677
val config = ConfigFactory.parseString("stateSelection {\n" +
7778
"inMemory {\n" +
7879
"cacheSize: 9000\n" +
79-
"indexingStrategies: [\"${VaultWatcherService.IndexingType.EXTERNAL_ID}\", \"${VaultWatcherService.IndexingType.PUBLIC_KEY}\"]\n" +
80+
"indexingStrategies: [\"${IndexingType.EXTERNAL_ID}\", \"${IndexingType.PUBLIC_KEY}\"]\n" +
8081
"}\n" +
8182
"}")
8283
val cordappConfig = TypesafeCordappConfig(config)
8384
val inMemoryConfig = InMemorySelectionConfig.parse(cordappConfig)
8485
assertThat(inMemoryConfig.cacheSize).isEqualTo(9000)
85-
assertThat(inMemoryConfig.indexingStrategies).isEqualTo(listOf(VaultWatcherService.IndexingType.EXTERNAL_ID, VaultWatcherService.IndexingType.PUBLIC_KEY))
86+
assertThat(inMemoryConfig.indexingStrategies).isEqualTo(listOf(IndexingType.EXTERNAL_ID, IndexingType.PUBLIC_KEY))
8687
}
8788

8889
@Test

0 commit comments

Comments
 (0)