StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

introduction of upstream update event

Browse Source
This commit is contained in:
a10zn8
2022-11-16 14:11:16 +04:00
parent 443839062f
commit 53e170b0af
17 changed files with 172 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
build.gradle
View File

@@ -289,3 +289,6 @@ detekt {
tasks.withType(Detekt).configureEach { tasks.withType(Detekt).configureEach {
jvmTarget = "13" jvmTarget = "13"
} }
// formats code for each build
tasks.findByName("ktlintCheck").dependsOn("ktlintFormat")

27
src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt
View File

@@ -1,6 +1,7 @@
package io.emeraldpay.dshackle.config.context package io.emeraldpay.dshackle.config.context
import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.upstream.CallTargetsHolder
import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
@@ -13,13 +14,31 @@ import org.springframework.context.annotation.Configuration
@Configuration @Configuration
class MultistreamsConfig { class MultistreamsConfig {
@Bean @Bean
fun allMultistreams(cachesFactory: CachesFactory): List<Multistream> { fun allMultistreams(
cachesFactory: CachesFactory,
callTargetsHolder: CallTargetsHolder
): List<Multistream> {
return Chain.values() return Chain.values()
.mapNotNull { chain -> .mapNotNull { chain ->
when (BlockchainType.from(chain)) { when (BlockchainType.from(chain)) {
BlockchainType.EVM_POS -> EthereumPosMultiStream(chain, ArrayList(), cachesFactory.getCaches(chain)) BlockchainType.EVM_POS -> EthereumPosMultiStream(
BlockchainType.EVM_POW -> EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) chain,
BlockchainType.BITCOIN -> BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) ArrayList(),
cachesFactory.getCaches(chain),
callTargetsHolder
)
BlockchainType.EVM_POW -> EthereumMultistream(
chain,
ArrayList(),
cachesFactory.getCaches(chain),
callTargetsHolder
)
BlockchainType.BITCOIN -> BitcoinMultistream(
chain,
ArrayList(),
cachesFactory.getCaches(chain),
callTargetsHolder
)
else -> null else -> null
} }
} }

30
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -21,13 +21,7 @@ import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HttpRpcFactory
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead
@@ -50,19 +44,20 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired import org.springframework.context.ApplicationEventPublisher
import org.springframework.stereotype.Repository import org.springframework.stereotype.Component
import java.net.URI import java.net.URI
import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicInteger
import java.util.function.Function import java.util.function.Function
import javax.annotation.PostConstruct import javax.annotation.PostConstruct
import kotlin.math.abs import kotlin.math.abs
@Repository @Component
open class ConfiguredUpstreams( open class ConfiguredUpstreams(
@Autowired private val currentUpstreams: CurrentMultistreamHolder, private val fileResolver: FileResolver,
@Autowired private val fileResolver: FileResolver, private val config: UpstreamsConfig,
@Autowired private val config: UpstreamsConfig private val callTargets: CallTargetsHolder,
private val eventPublisher: ApplicationEventPublisher
) { ) {
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
@@ -115,7 +110,8 @@ open class ConfiguredUpstreams(
} }
} }
upstream?.let { upstream?.let {
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED)) val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED)
eventPublisher.publishEvent(event)
} }
} }
} }
@@ -150,7 +146,7 @@ open class ConfiguredUpstreams(
fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods { fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
return if (config.methods != null) { return if (config.methods != null) {
ManagedCallMethods( ManagedCallMethods(
currentUpstreams.getDefaultMethods(chain), callTargets.getDefaultMethods(chain),
config.methods!!.enabled.map { it.name }.toSet(), config.methods!!.enabled.map { it.name }.toSet(),
config.methods!!.disabled.map { it.name }.toSet() config.methods!!.disabled.map { it.name }.toSet()
).also { ).also {
@@ -164,7 +160,7 @@ open class ConfiguredUpstreams(
} }
} }
} else { } else {
currentUpstreams.getDefaultMethods(chain) callTargets.getDefaultMethods(chain)
} }
} }
@@ -315,7 +311,7 @@ open class ConfiguredUpstreams(
.doOnNext { .doOnNext {
log.info("Chain ${it.chain} ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}. With caps: ${it.upstream.getCapabilities()}") log.info("Chain ${it.chain} ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}. With caps: ${it.upstream.getCapabilities()}")
} }
.subscribe(currentUpstreams::update) .subscribe(eventPublisher::publishEvent)
} }
private fun buildHttpFactory(conn: UpstreamsConfig.RpcConnection, urls: ArrayList<URI>? = null): HttpRpcFactory? { private fun buildHttpFactory(conn: UpstreamsConfig.RpcConnection, urls: ArrayList<URI>? = null): HttpRpcFactory? {

2
src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt → src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt
View File

@@ -24,7 +24,7 @@ import io.emeraldpay.grpc.Chain
/** /**
* An update event to the list of currently available upstreams. * An update event to the list of currently available upstreams.
*/ */
class UpstreamChange( class UpstreamChangeEvent(
/** /**
* Target blockchain * Target blockchain
*/ */

29
src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt Normal file
View File

@@ -0,0 +1,29 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.springframework.stereotype.Component
import java.util.HashMap
@Component
class CallTargetsHolder {
private val callTargets = HashMap<Chain, CallMethods>()
fun getDefaultMethods(chain: Chain): CallMethods {
return callTargets[chain] ?: return setupDefaultMethods(chain)
}
private fun setupDefaultMethods(chain: Chain): CallMethods {
val created = when (BlockchainType.from(chain)) {
BlockchainType.EVM_POW -> DefaultEthereumMethods(chain)
BlockchainType.BITCOIN -> DefaultBitcoinMethods()
BlockchainType.EVM_POS -> DefaultEthereumMethods(chain)
else -> throw IllegalStateException("Unsupported chain: $chain")
}
callTargets[chain] = created
return created
}
}

79
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -16,15 +16,6 @@
*/ */
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component import org.springframework.stereotype.Component
@@ -49,61 +40,8 @@ open class CurrentMultistreamHolder(
private val chainsBus = Sinks.many() private val chainsBus = Sinks.many()
.multicast() .multicast()
.directBestEffort<Chain>() .directBestEffort<Chain>()
private val callTargets = HashMap<Chain, CallMethods>()
private val updateLock = ReentrantLock() private val updateLock = ReentrantLock()
fun update(change: UpstreamChange) {
updateLock.withLock {
log.debug("Upstream update: ${change.type} ${change.chain} via ${change.upstream.getId()}")
val chain = change.chain
try {
when (BlockchainType.from(chain)) {
BlockchainType.EVM_POW -> {
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping.getValue(chain)
processUpdate(change, up, current)
}
BlockchainType.EVM_POS -> {
val up = change.upstream.cast(EthereumPosUpstream::class.java)
val current = chainMapping.getValue(chain)
processUpdate(change, up, current)
}
BlockchainType.BITCOIN -> {
val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping.getValue(chain)
processUpdate(change, up, current)
}
else -> {
log.error("Update for unsupported chain: $chain")
}
}
} catch (e: Throwable) {
log.error("Failed to update upstream", e)
}
}
}
fun processUpdate(change: UpstreamChange, up: Upstream, current: Multistream) {
val chain = change.chain
if (change.type == UpstreamChange.ChangeType.REMOVED) {
current.removeUpstream(up.getId())
log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
} else {
if (up is CachesEnabled) {
up.setCaches(current.caches)
}
current.addUpstream(up)
if (!callTargets.containsKey(chain)) {
setupDefaultMethods(chain)
}
log.info("Upstream ${change.upstream.getId()} with chain $chain has been added")
}
}
override fun getUpstream(chain: Chain): Multistream? { override fun getUpstream(chain: Chain): Multistream? {
return chainMapping[chain] return chainMapping[chain]
} }
@@ -122,23 +60,8 @@ open class CurrentMultistreamHolder(
) )
} }
override fun getDefaultMethods(chain: Chain): CallMethods {
return callTargets[chain] ?: return setupDefaultMethods(chain)
}
fun setupDefaultMethods(chain: Chain): CallMethods {
val created = when (BlockchainType.from(chain)) {
BlockchainType.EVM_POW -> DefaultEthereumMethods(chain)
BlockchainType.BITCOIN -> DefaultBitcoinMethods()
BlockchainType.EVM_POS -> DefaultEthereumMethods(chain)
else -> throw IllegalStateException("Unsupported chain: $chain")
}
callTargets[chain] = created
return created
}
override fun isAvailable(chain: Chain): Boolean { override fun isAvailable(chain: Chain): Boolean {
return chainMapping.containsKey(chain) && callTargets.containsKey(chain) return chainMapping.getValue(chain).isAvailable()
} }
@PreDestroy @PreDestroy

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -17,8 +17,10 @@
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -30,6 +32,7 @@ import org.apache.commons.collections4.Factory
import org.apache.commons.collections4.FunctorException import org.apache.commons.collections4.FunctorException
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
import org.springframework.context.event.EventListener
import reactor.core.Disposable import reactor.core.Disposable
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
@@ -48,7 +51,8 @@ abstract class Multistream(
val chain: Chain, val chain: Chain,
private val upstreams: MutableList<Upstream>, private val upstreams: MutableList<Upstream>,
val caches: Caches, val caches: Caches,
val postprocessor: RequestPostprocessor val postprocessor: RequestPostprocessor,
val callTargetsHolder: CallTargetsHolder
) : Upstream, Lifecycle { ) : Upstream, Lifecycle {
companion object { companion object {
@@ -316,6 +320,23 @@ abstract class Multistream(
log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak]") log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak]")
} }
@EventListener
fun onUpstreamChange(event: UpstreamChangeEvent) {
val chain = event.chain
if (this.chain == chain) {
if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) {
removeUpstream(event.upstream.getId())
log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
} else {
if (event.upstream is CachesEnabled) {
event.upstream.setCaches(caches)
}
addUpstream(event.upstream)
log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
}
}
}
// -------------------------------------------------------------------------------------------------------- // --------------------------------------------------------------------------------------------------------
class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt
View File

@@ -16,7 +16,6 @@
*/ */
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
@@ -27,6 +26,5 @@ interface MultistreamHolder {
fun getUpstream(chain: Chain): Multistream? fun getUpstream(chain: Chain): Multistream?
fun getAvailable(): List<Chain> fun getAvailable(): List<Chain>
fun observeChains(): Flux<Chain> fun observeChains(): Flux<Chain>
fun getDefaultMethods(chain: Chain): CallMethods
fun isAvailable(chain: Chain): Boolean fun isAvailable(chain: Chain): Boolean
} }

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -18,14 +18,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.RequestPostprocessor
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -39,8 +32,9 @@ import reactor.core.publisher.Mono
open class BitcoinMultistream( open class BitcoinMultistream(
chain: Chain, chain: Chain,
private val sourceUpstreams: MutableList<BitcoinUpstream>, private val sourceUpstreams: MutableList<BitcoinUpstream>,
caches: Caches caches: Caches,
) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty()), Lifecycle { callTargetsHolder: CallTargetsHolder
) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty(), callTargetsHolder), Lifecycle {
companion object { companion object {
private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -20,13 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -42,8 +36,9 @@ import reactor.core.publisher.Mono
open class EthereumMultistream( open class EthereumMultistream(
chain: Chain, chain: Chain,
val upstreams: MutableList<EthereumUpstream>, val upstreams: MutableList<EthereumUpstream>,
caches: Caches caches: Caches,
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)), EthereumLikeMultistream { callTargetsHolder: CallTargetsHolder
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches), callTargetsHolder), EthereumLikeMultistream {
companion object { companion object {
private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
View File

@@ -20,13 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -42,8 +36,9 @@ import reactor.core.publisher.Mono
open class EthereumPosMultiStream( open class EthereumPosMultiStream(
chain: Chain, chain: Chain,
val upstreams: MutableList<EthereumPosUpstream>, val upstreams: MutableList<EthereumPosUpstream>,
caches: Caches caches: Caches,
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)), EthereumLikeMultistream { callTargetsHolder: CallTargetsHolder
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches), callTargetsHolder), EthereumLikeMultistream {
companion object { companion object {
private val log = LoggerFactory.getLogger(EthereumPosMultiStream::class.java) private val log = LoggerFactory.getLogger(EthereumPosMultiStream::class.java)

30
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
@@ -69,7 +69,7 @@ class GrpcUpstreams(
private val known = HashMap<Chain, DefaultUpstream>() private val known = HashMap<Chain, DefaultUpstream>()
private val lock = ReentrantLock() private val lock = ReentrantLock()
fun start(): Flux<UpstreamChange> { fun start(): Flux<UpstreamChangeEvent> {
val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) { val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) {
NettyChannelBuilder.forAddress(host, port) NettyChannelBuilder.forAddress(host, port)
// some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces)
@@ -122,7 +122,7 @@ class GrpcUpstreams(
return updates return updates
} }
fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux<UpstreamChange> { fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux<UpstreamChangeEvent> {
val current = value.chainsList.filter { val current = value.chainsList.filter {
Chain.byId(it.chain.number) != Chain.UNSPECIFIED Chain.byId(it.chain.number) != Chain.UNSPECIFIED
}.mapNotNull { chainDetails -> }.mapNotNull { chainDetails ->
@@ -138,14 +138,14 @@ class GrpcUpstreams(
} }
val added = current.filter { val added = current.filter {
it.type == UpstreamChange.ChangeType.ADDED it.type == UpstreamChangeEvent.ChangeType.ADDED
} }
val removed = known.filterNot { kv -> val removed = known.filterNot { kv ->
val stillCurrent = current.any { c -> c.chain == kv.key } val stillCurrent = current.any { c -> c.chain == kv.key }
stillCurrent stillCurrent
}.map { }.map {
UpstreamChange(it.key, known.remove(it.key)!!, UpstreamChange.ChangeType.REMOVED) UpstreamChangeEvent(it.key, known.remove(it.key)!!, UpstreamChangeEvent.ChangeType.REMOVED)
} }
return Flux.fromIterable(removed + added) return Flux.fromIterable(removed + added)
} }
@@ -172,7 +172,7 @@ class GrpcUpstreams(
return sslContext.build() return sslContext.build()
} }
fun getOrCreate(chain: Chain): UpstreamChange { fun getOrCreate(chain: Chain): UpstreamChangeEvent {
val metricsTags = listOf( val metricsTags = listOf(
Tag.of("upstream", id), Tag.of("upstream", id),
Tag.of("chain", chain.chainCode) Tag.of("chain", chain.chainCode)
@@ -202,7 +202,7 @@ class GrpcUpstreams(
} }
} }
fun getOrCreateEthereum(chain: Chain, metrics: RpcMetrics): UpstreamChange { fun getOrCreateEthereum(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent {
lock.withLock { lock.withLock {
val current = known[chain] val current = known[chain]
return if (current == null) { return if (current == null) {
@@ -211,14 +211,14 @@ class GrpcUpstreams(
created.timeout = this.timeout created.timeout = this.timeout
known[chain] = created known[chain] = created
created.start() created.start()
UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED)
} else { } else {
UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED) UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED)
} }
} }
} }
fun getOrCreateEthereumPos(chain: Chain, metrics: RpcMetrics): UpstreamChange { fun getOrCreateEthereumPos(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent {
lock.withLock { lock.withLock {
val current = known[chain] val current = known[chain]
return if (current == null) { return if (current == null) {
@@ -227,14 +227,14 @@ class GrpcUpstreams(
created.timeout = this.timeout created.timeout = this.timeout
known[chain] = created known[chain] = created
created.start() created.start()
UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED)
} else { } else {
UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED) UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED)
} }
} }
} }
fun getOrCreateBitcoin(chain: Chain, metrics: RpcMetrics): UpstreamChange { fun getOrCreateBitcoin(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent {
lock.withLock { lock.withLock {
val current = known[chain] val current = known[chain]
return if (current == null) { return if (current == null) {
@@ -243,9 +243,9 @@ class GrpcUpstreams(
created.timeout = this.timeout created.timeout = this.timeout
known[chain] = created known[chain] = created
created.start() created.start()
UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED)
} else { } else {
UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED) UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED)
} }
} }
} }

34
src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy
View File

@@ -4,21 +4,24 @@ import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.quorum.NonEmptyQuorum import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
import io.emeraldpay.dshackle.upstream.CallTargetsHolder
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import org.springframework.context.ApplicationEventPublisher
import spock.lang.Specification import spock.lang.Specification
class ConfiguredUpstreamsSpec extends Specification { class ConfiguredUpstreamsSpec extends Specification {
def "Applied quorum to extra methods"() { def "Applied quorum to extra methods"() {
setup: setup:
def currentUpstreams = Mock(CurrentMultistreamHolder) { def callTargetsHolder = new CallTargetsHolder()
_ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM)
}
def configurer = new ConfiguredUpstreams( def configurer = new ConfiguredUpstreams(
currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig) Stub(FileResolver),
Stub(UpstreamsConfig),
callTargetsHolder,
Mock(ApplicationEventPublisher)
) )
def methods = new UpstreamsConfig.Methods( def methods = new UpstreamsConfig.Methods(
[ [
@@ -38,11 +41,12 @@ class ConfiguredUpstreamsSpec extends Specification {
def "Got static response from extra methods"() { def "Got static response from extra methods"() {
setup: setup:
def currentUpstreams = Mock(CurrentMultistreamHolder) { def callTargetsHolder = new CallTargetsHolder()
_ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM)
}
def configurer = new ConfiguredUpstreams( def configurer = new ConfiguredUpstreams(
currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig) Stub(FileResolver),
Stub(UpstreamsConfig),
callTargetsHolder,
Mock(ApplicationEventPublisher)
) )
def methods = new UpstreamsConfig.Methods( def methods = new UpstreamsConfig.Methods(
[ [
@@ -61,7 +65,12 @@ class ConfiguredUpstreamsSpec extends Specification {
def "Calculate node-id"() { def "Calculate node-id"() {
setup: setup:
def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig) def callTargetsHolder = new CallTargetsHolder()
def configurer = new ConfiguredUpstreams(
Stub(FileResolver),
Stub(UpstreamsConfig),
callTargetsHolder,
Mock(ApplicationEventPublisher)
) )
expect: expect:
configurer.getHash(node, src) == expected configurer.getHash(node, src) == expected
@@ -75,7 +84,12 @@ class ConfiguredUpstreamsSpec extends Specification {
def "Calculate node-id conflicting results"() { def "Calculate node-id conflicting results"() {
setup: setup:
def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig) def callTargetsHolder = new CallTargetsHolder()
def configurer = new ConfiguredUpstreams(
Stub(FileResolver),
Stub(UpstreamsConfig),
callTargetsHolder,
Mock(ApplicationEventPublisher)
) )
when: when:
def h1 = configurer.getHash(null, "hohoho") def h1 = configurer.getHash(null, "hohoho")

15
src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
View File

@@ -50,7 +50,7 @@ class MultistreamHolderMock implements MultistreamHolder {
if (up instanceof EthereumPosMultiStream) { if (up instanceof EthereumPosMultiStream) {
upstreams[chain] = up upstreams[chain] = up
} else if (up instanceof EthereumPosRpcUpstream) { } else if (up instanceof EthereumPosRpcUpstream) {
upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default()) upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default(), TestingCommons.callTargetsHolder)
} else { } else {
throw new IllegalArgumentException("Unsupported upstream type ${up.class}") throw new IllegalArgumentException("Unsupported upstream type ${up.class}")
} }
@@ -59,7 +59,7 @@ class MultistreamHolderMock implements MultistreamHolder {
if (up instanceof BitcoinMultistream) { if (up instanceof BitcoinMultistream) {
upstreams[chain] = up upstreams[chain] = up
} else if (up instanceof BitcoinRpcUpstream) { } else if (up instanceof BitcoinRpcUpstream) {
upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinRpcUpstream], Caches.default()) upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinRpcUpstream], Caches.default(), TestingCommons.callTargetsHolder)
} else { } else {
throw new IllegalArgumentException("Unsupported upstream type ${up.class}") throw new IllegalArgumentException("Unsupported upstream type ${up.class}")
} }
@@ -86,15 +86,6 @@ class MultistreamHolderMock implements MultistreamHolder {
return Flux.fromIterable(getAvailable()) return Flux.fromIterable(getAvailable())
} }
@Override
DefaultEthereumMethods getDefaultMethods(@NotNull Chain chain) {
if (target[chain] == null) {
DefaultEthereumMethods targets = new DefaultEthereumMethods(chain)
target[chain] = targets
}
return target[chain]
}
@Override @Override
boolean isAvailable(@NotNull Chain chain) { boolean isAvailable(@NotNull Chain chain) {
return upstreams.containsKey(chain) return upstreams.containsKey(chain)
@@ -107,7 +98,7 @@ class MultistreamHolderMock implements MultistreamHolder {
Head customHead = null Head customHead = null
EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams, @NotNull Caches caches) { EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches) super(chain, upstreams, caches, TestingCommons.callTargetsHolder)
} }
EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams) { EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams) {

14
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

@@ -24,8 +24,10 @@ import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.CallTargetsHolder
import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -78,7 +80,7 @@ class TestingCommons {
} }
static Multistream multistream(EthereumPosRpcUpstreamMock up) { static Multistream multistream(EthereumPosRpcUpstreamMock up) {
return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default()).tap { return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), callTargetsHolder).tap {
start() start()
} }
} }
@@ -94,12 +96,16 @@ class TestingCommons {
static List<Multistream> defaultMultistreams() { static List<Multistream> defaultMultistreams() {
return [ return [
multistreamWithoutUpstreams(Chain.ETHEREUM), multistreamWithoutUpstreams(Chain.ETHEREUM),
multistreamWithoutUpstreams(Chain.ETHEREUM_CLASSIC) multistreamClassicWithoutUpstreams(Chain.ETHEREUM_CLASSIC)
] ]
} }
static Multistream multistreamWithoutUpstreams(Chain chain) { static Multistream multistreamWithoutUpstreams(Chain chain) {
return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain)) return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), callTargetsHolder)
}
static Multistream multistreamClassicWithoutUpstreams(Chain chain) {
return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), callTargetsHolder)
} }
static FileResolver fileResolver() { static FileResolver fileResolver() {
@@ -138,4 +144,6 @@ class TestingCommons {
} }
static MeterRegistry meterRegistry = new LoggingMeterRegistry() static MeterRegistry meterRegistry = new LoggingMeterRegistry()
static CallTargetsHolder callTargetsHolder = new CallTargetsHolder()
} }

21
src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy
View File

@@ -15,7 +15,7 @@
*/ */
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock
import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
@@ -29,7 +29,7 @@ class CurrentMultistreamHolderSpec extends Specification {
def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams()) def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams())
def up = new EthereumPosRpcUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api()) def up = new EthereumPosRpcUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api())
when: when:
current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up, UpstreamChangeEvent.ChangeType.ADDED))
then: then:
current.getAvailable() == [Chain.ETHEREUM] current.getAvailable() == [Chain.ETHEREUM]
current.getUpstream(Chain.ETHEREUM).getAll()[0] == up current.getUpstream(Chain.ETHEREUM).getAll()[0] == up
@@ -42,9 +42,10 @@ class CurrentMultistreamHolderSpec extends Specification {
def up2 = new EthereumRpcUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up2 = new EthereumRpcUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api())
def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api())
when: when:
current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1, UpstreamChangeEvent.ChangeType.ADDED))
current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM_CLASSIC).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC, up2, UpstreamChangeEvent.ChangeType.ADDED))
current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up3, UpstreamChangeEvent.ChangeType.ADDED))
current.getUpstream(Chain.ETHEREUM_CLASSIC).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up3, UpstreamChangeEvent.ChangeType.ADDED))
then: then:
current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet() current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet()
current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up1, up3].toSet() current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up1, up3].toSet()
@@ -59,10 +60,10 @@ class CurrentMultistreamHolderSpec extends Specification {
def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api())
def up1_del = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up1_del = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
when: when:
current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1, UpstreamChangeEvent.ChangeType.ADDED))
current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM_CLASSIC).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC, up2, UpstreamChangeEvent.ChangeType.ADDED))
current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up3, UpstreamChangeEvent.ChangeType.ADDED))
current.update(new UpstreamChange(Chain.ETHEREUM, up1_del, UpstreamChange.ChangeType.REMOVED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1_del, UpstreamChangeEvent.ChangeType.REMOVED))
then: then:
current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet() current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet()
current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up3].toSet() current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up3].toSet()
@@ -80,7 +81,7 @@ class CurrentMultistreamHolderSpec extends Specification {
!act !act
when: when:
current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1, UpstreamChangeEvent.ChangeType.ADDED))
act = current.isAvailable(Chain.ETHEREUM) act = current.isAvailable(Chain.ETHEREUM)
then: then:

8
src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
View File

@@ -50,7 +50,7 @@ class MultistreamSpec extends Specification {
setup: setup:
def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
def up2 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) def up2 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"]))
def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default()) def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), TestingCommons.callTargetsHolder)
when: when:
aggr.onUpstreamsUpdated() aggr.onUpstreamsUpdated()
def act = aggr.getMethods() def act = aggr.getMethods()
@@ -206,7 +206,7 @@ class MultistreamSpec extends Specification {
def up1 = TestingCommons.upstream("test-1", "internal") def up1 = TestingCommons.upstream("test-1", "internal")
def up2 = TestingCommons.upstream("test-2", "external") def up2 = TestingCommons.upstream("test-2", "external")
def up3 = TestingCommons.upstream("test-3", "external") def up3 = TestingCommons.upstream("test-3", "external")
def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default()) def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), TestingCommons.callTargetsHolder)
expect: expect:
multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock)
@@ -345,7 +345,7 @@ class MultistreamSpec extends Specification {
class TestMultistream extends Multistream { class TestMultistream extends Multistream {
TestMultistream(List<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) { TestMultistream(List<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) {
super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor) super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor, TestingCommons.callTargetsHolder)
} }
@Override @Override
@@ -386,7 +386,7 @@ class MultistreamSpec extends Specification {
class TestEthereumPosMultistream extends EthereumPosMultiStream { class TestEthereumPosMultistream extends EthereumPosMultiStream {
TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumPosUpstream> upstreams, @NotNull Caches caches) { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches) super(chain, upstreams, caches, TestingCommons.callTargetsHolder)
} }
@Override @Override