diff --git a/build.gradle b/build.gradle index 41a94e9d..eb15c928 100644 --- a/build.gradle +++ b/build.gradle @@ -289,3 +289,6 @@ detekt { tasks.withType(Detekt).configureEach { jvmTarget = "13" } + +// formats code for each build +tasks.findByName("ktlintCheck").dependsOn("ktlintFormat") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Starter.kt b/src/main/kotlin/io/emeraldpay/dshackle/Starter.kt index 807b399b..525eda35 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Starter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Starter.kt @@ -20,12 +20,10 @@ import org.slf4j.LoggerFactory import org.springframework.boot.ResourceBanner import org.springframework.boot.SpringApplication import org.springframework.boot.autoconfigure.SpringBootApplication -import org.springframework.context.annotation.Import import org.springframework.core.io.ClassPathResource import org.springframework.core.io.support.ResourcePropertySource @SpringBootApplication(scanBasePackages = ["io.emeraldpay.dshackle"]) -@Import(Config::class) open class Starter private val log = LoggerFactory.getLogger(Starter::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt new file mode 100644 index 00000000..3329f93e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -0,0 +1,78 @@ +package io.emeraldpay.dshackle.config.context + +import io.emeraldpay.dshackle.cache.CachesFactory +import io.emeraldpay.dshackle.upstream.CallTargetsHolder +import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream +import io.emeraldpay.grpc.BlockchainType +import io.emeraldpay.grpc.Chain +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration + +@Configuration +open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) { + @Bean + open fun allMultistreams( + cachesFactory: CachesFactory, + callTargetsHolder: CallTargetsHolder + ): List { + return Chain.values() + .filterNot { it == Chain.UNSPECIFIED } + .mapNotNull { chain -> + when (BlockchainType.from(chain)) { + BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory) + BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory) + BlockchainType.BITCOIN -> bitcoinMultistream(chain, cachesFactory) + else -> null + } + } + } + + private fun ethereumMultistream( + chain: Chain, + cachesFactory: CachesFactory + ): EthereumMultistream { + val name = "multi-ethereum-$chain" + + return EthereumMultistream( + chain, + ArrayList(), + cachesFactory.getCaches(chain) + ).also { register(it, name) } + } + + open fun ethereumPosMultistream( + chain: Chain, + cachesFactory: CachesFactory + ): EthereumPosMultiStream { + val name = "multi-ethereum-pos-$chain" + + return EthereumPosMultiStream( + chain, + ArrayList(), + cachesFactory.getCaches(chain) + ).also { register(it, name) } + } + + open fun bitcoinMultistream( + chain: Chain, + cachesFactory: CachesFactory + ): BitcoinMultistream { + val name = "multi-bitcoin-$chain" + + return BitcoinMultistream( + chain, + ArrayList(), + cachesFactory.getCaches(chain) + ).also { register(it, name) } + } + + private fun register(bean: Any, name: String) { + beanFactory.initializeBean(bean, name) + beanFactory.autowireBean(bean) + this.beanFactory.registerSingleton(name, bean) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 956b187d..93926399 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -26,11 +26,10 @@ import io.emeraldpay.dshackle.quorum.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.startup.ConfiguredUpstreams -import io.emeraldpay.dshackle.upstream.ApiSource -import io.emeraldpay.dshackle.upstream.Multistream -import io.emeraldpay.dshackle.upstream.MultistreamHolder -import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError @@ -45,7 +44,7 @@ import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.Metrics import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Autowired +import org.springframework.context.event.EventListener import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -55,9 +54,9 @@ import java.util.concurrent.atomic.AtomicInteger @Service open class NativeCall( - @Autowired private val multistreamHolder: MultistreamHolder, - @Autowired private val configuredUpstreams: ConfiguredUpstreams, - @Autowired private val signer: ResponseSigner + private val multistreamHolder: MultistreamHolder, + private val configuredUpstreams: ConfiguredUpstreams, + private val signer: ResponseSigner ) { private val log = LoggerFactory.getLogger(NativeCall::class.java) @@ -66,18 +65,19 @@ open class NativeCall( var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() private val ethereumCallSelectors = EnumMap(Chain::class.java) - init { - val casting = mapOf( + companion object { + val casting: Map> = mapOf( BlockchainType.EVM_POS to EthereumPosMultiStream::class.java, - BlockchainType.EVM_POW to EthereumMultistream::class.java, + BlockchainType.EVM_POW to EthereumMultistream::class.java ) + } - multistreamHolder.observeChains().subscribe { chain -> - casting[BlockchainType.from(chain)]?.let { cast -> - multistreamHolder.getUpstream(chain)?.let { up -> - val reader = up.cast(cast).getReader() - ethereumCallSelectors.putIfAbsent(chain, EthereumCallSelector(reader.heightByHash())) - } + @EventListener + fun onUpstreamChangeEvent(event: UpstreamChangeEvent) { + casting[BlockchainType.from(event.chain)]?.let { cast -> + multistreamHolder.getUpstream(event.chain)?.let { up -> + val reader = up.cast(cast).getReader() + ethereumCallSelectors.putIfAbsent(event.chain, EthereumCallSelector(reader.heightByHash())) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt index 0536d655..d64b493c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt @@ -20,6 +20,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.SilentException +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.dshackle.upstream.Selector @@ -33,12 +34,12 @@ import org.bitcoinj.params.MainNetParams import org.bitcoinj.params.TestNet3Params import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired +import org.springframework.context.event.EventListener import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger import java.util.concurrent.ConcurrentHashMap -import javax.annotation.PostConstruct @Service class TrackBitcoinAddress( @@ -67,15 +68,13 @@ class TrackBitcoinAddress( Selector.CapabilityMatcher(Capability.BALANCE) ) - @PostConstruct - fun listenChains() { - multistreamHolder.observeChains().subscribe { chain -> - multistreamHolder.getUpstream(chain)?.let { mup -> - val available = mup.getAll().any { up -> - !up.isGrpc() && up.getCapabilities().contains(Capability.BALANCE) - } - setBalanceAvailability(chain, available) + @EventListener + fun onUpstreamChangeEvent(event: UpstreamChangeEvent) { + multistreamHolder.getUpstream(event.chain)?.let { mup -> + val available = mup.getAll().any { up -> + !up.isGrpc() && up.getCapabilities().contains(Capability.BALANCE) } + setBalanceAvailability(event.chain, available) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 3a86c6ac..51aed124 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -21,13 +21,7 @@ import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.BlockValidator -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.* import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead @@ -50,28 +44,29 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Autowired -import org.springframework.stereotype.Repository +import org.springframework.boot.ApplicationArguments +import org.springframework.boot.ApplicationRunner +import org.springframework.context.ApplicationEventPublisher +import org.springframework.stereotype.Component import java.net.URI import java.util.concurrent.atomic.AtomicInteger import java.util.function.Function -import javax.annotation.PostConstruct import kotlin.math.abs -@Repository +@Component open class ConfiguredUpstreams( - @Autowired private val currentUpstreams: CurrentMultistreamHolder, - @Autowired private val fileResolver: FileResolver, - @Autowired private val config: UpstreamsConfig -) { + private val fileResolver: FileResolver, + private val config: UpstreamsConfig, + private val callTargets: CallTargetsHolder, + private val eventPublisher: ApplicationEventPublisher +) : ApplicationRunner { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) private var seq = AtomicInteger(0) private val hashes: MutableMap = HashMap() - @PostConstruct - fun start() { + override fun run(args: ApplicationArguments) { log.debug("Starting upstreams") val defaultOptions = buildDefaultOptions(config) config.upstreams.forEach { up -> @@ -115,7 +110,8 @@ open class ConfiguredUpstreams( } } 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 { return if (config.methods != null) { ManagedCallMethods( - currentUpstreams.getDefaultMethods(chain), + callTargets.getDefaultMethods(chain), config.methods!!.enabled.map { it.name }.toSet(), config.methods!!.disabled.map { it.name }.toSet() ).also { @@ -164,7 +160,7 @@ open class ConfiguredUpstreams( } } } else { - currentUpstreams.getDefaultMethods(chain) + callTargets.getDefaultMethods(chain) } } @@ -315,7 +311,7 @@ open class ConfiguredUpstreams( .doOnNext { 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? = null): HttpRpcFactory? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt similarity index 98% rename from src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt rename to src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt index 213e55a1..34bcc8bf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt @@ -24,7 +24,7 @@ import io.emeraldpay.grpc.Chain /** * An update event to the list of currently available upstreams. */ -class UpstreamChange( +class UpstreamChangeEvent( /** * Target blockchain */ diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt new file mode 100644 index 00000000..2ad03f39 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt @@ -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() + + 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 + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index 7587d6c2..54a152b3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -16,156 +16,40 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.cache.CachesEnabled -import io.emeraldpay.dshackle.cache.CachesFactory -import io.emeraldpay.dshackle.startup.UpstreamChange -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream -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.EthereumMultistream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream -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 org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Autowired -import org.springframework.stereotype.Repository -import reactor.core.publisher.Flux -import reactor.core.publisher.Sinks -import java.util.Collections -import java.util.concurrent.Callable -import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.locks.ReentrantLock +import org.springframework.stereotype.Component import javax.annotation.PreDestroy -import kotlin.concurrent.withLock -@Repository +@Component open class CurrentMultistreamHolder( - @Autowired private val cachesFactory: CachesFactory + multistreams: List ) : MultistreamHolder { private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java) - private val chainMapping = ConcurrentHashMap() - private val chainsBus = Sinks.many() - .multicast() - .directBestEffort() - private val callTargets = HashMap() - 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[chain] - val factory = Callable { - EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) - } - processUpdate(change, up, current, factory) - } - BlockchainType.EVM_POS -> { - val up = change.upstream.cast(EthereumPosUpstream::class.java) - val current = chainMapping[chain] - val factory = Callable { - EthereumPosMultiStream(chain, ArrayList(), cachesFactory.getCaches(chain)) - } - processUpdate(change, up, current, factory) - } - BlockchainType.BITCOIN -> { - val up = change.upstream.cast(BitcoinUpstream::class.java) - val current = chainMapping[chain] - val factory = Callable { - BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) - } - processUpdate(change, up, current, factory) - } - 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?, factory: Callable) { - 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 (current == null) { - val created = factory.call() - if (up is CachesEnabled) { - up.setCaches(created.caches) - } - created.addUpstream(up) - created.start() - chainMapping[chain] = created - chainsBus.tryEmitNext(chain) - } 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") - } - } + private val chainMapping = multistreams.associateBy { it.chain } override fun getUpstream(chain: Chain): Multistream? { return chainMapping[chain] } override fun getAvailable(): List { - return Collections.unmodifiableList(chainMapping.keys.toList()) - } - - override fun observeChains(): Flux { - return Flux.concat( - Flux.fromIterable(getAvailable()), - chainsBus.asFlux() - ) - } - - 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 + return chainMapping.values.asSequence() + .filter { it.haveUpstreams() } + .map { it.chain } + .toList() } override fun isAvailable(chain: Chain): Boolean { - return chainMapping.containsKey(chain) && callTargets.containsKey(chain) + return chainMapping.getValue(chain).isAvailable() } @PreDestroy fun shutdown() { log.info("Closing upstream connections...") - updateLock.withLock { - chainMapping.values.forEach { - it.stop() - } - chainMapping.clear() + chainMapping.values.forEach { + it.stop() } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Lifecycle.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Lifecycle.kt new file mode 100644 index 00000000..ff7eb41d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Lifecycle.kt @@ -0,0 +1,7 @@ +package io.emeraldpay.dshackle.upstream + +interface Lifecycle { + fun start() + fun stop() + fun isRunning(): Boolean +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt index 42e654e0..7ea7d7f8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt @@ -21,7 +21,6 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -43,7 +42,7 @@ class MergedHead( override fun start() { super.start() sources.forEach { head -> - if (head is Lifecycle && !head.isRunning) { + if (head is Lifecycle && !head.isRunning()) { head.start() } } @@ -56,7 +55,7 @@ class MergedHead( override fun stop() { super.stop() sources.forEach { head -> - if (head is Lifecycle && head.isRunning) { + if (head is Lifecycle && head.isRunning()) { head.stop() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index c04dd9e0..b6de244c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -17,8 +17,10 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig 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.CallMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -29,13 +31,15 @@ import io.micrometer.core.instrument.Tag import org.apache.commons.collections4.Factory import org.apache.commons.collections4.FunctorException import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle +import org.springframework.context.event.EventListener +import org.springframework.core.Ordered +import org.springframework.core.annotation.Order import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.time.Duration import java.time.Instant -import java.util.Locale +import java.util.* import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock import java.util.function.Predicate @@ -56,6 +60,8 @@ abstract class Multistream( private const val metrics = "upstreams" } + private var started = false + private var cacheSubscription: Disposable? = null private val reconfigLock = ReentrantLock() private var callMethods: CallMethods? = null @@ -231,6 +237,7 @@ abstract class Multistream( // print status _change_ every 15 seconds, at most; otherwise prints it on interval of 30 seconds .sample(Duration.ofSeconds(15)) .subscribe { printStatus() } + started = true } override fun stop() { @@ -244,6 +251,7 @@ abstract class Multistream( } } lagObserver?.stop() + started = false } fun onHeadUpdated(head: Head) { @@ -316,6 +324,34 @@ abstract class Multistream( log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak]") } + fun test(event: UpstreamChangeEvent): Boolean { + return event.chain == this.chain + } + + @EventListener + @Order(Ordered.HIGHEST_PRECEDENCE) + fun onUpstreamChange(event: UpstreamChangeEvent) { + val chain = event.chain + if (this.chain == chain) { + if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { + removeUpstream(event.upstream.getId()) + log.error("Upstream ${event.upstream.getId()} with chain $chain has been removed") + } else { + if (event.upstream is CachesEnabled) { + event.upstream.setCaches(caches) + } + addUpstream(event.upstream) + if (!started) { + start() + } + log.error("Upstream ${event.upstream.getId()} with chain $chain has been added") + } + } + } + + fun haveUpstreams(): Boolean = + upstreams.isNotEmpty() + // -------------------------------------------------------------------------------------------------------- class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt index 852ffdee..80b4e564 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt @@ -16,9 +16,7 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.grpc.Chain -import reactor.core.publisher.Flux /** * Holds Multistreams configured for a chain. @@ -26,7 +24,5 @@ import reactor.core.publisher.Flux interface MultistreamHolder { fun getUpstream(chain: Chain): Multistream? fun getAvailable(): List - fun observeChains(): Flux - fun getDefaultMethods(chain: Chain): CallMethods fun isAvailable(chain: Chain): Boolean } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index d72a3fb5..aca3d567 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -18,21 +18,14 @@ package io.emeraldpay.dshackle.upstream.bitcoin import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.ChainFees -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.* +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.publisher.Mono @Suppress("UNCHECKED_CAST") @@ -142,7 +135,7 @@ open class BitcoinMultistream( } override fun isRunning(): Boolean { - return super.isRunning() || reader.isRunning + return super.isRunning() || reader.isRunning() } override fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt index f9f0496c..8ae68e9f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt @@ -19,13 +19,13 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.bitcoinj.core.Address import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.cast @@ -72,7 +72,7 @@ open class BitcoinReader( } override fun isRunning(): Boolean { - return mempool.isRunning + return mempool.isRunning() } override fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt index 72f500bc..5c018104 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt @@ -19,11 +19,11 @@ import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import org.springframework.scheduling.concurrent.CustomizableThreadFactory import reactor.core.Disposable import reactor.core.publisher.Flux diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt index e5ab2d29..23b82f35 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -27,7 +28,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable open class BitcoinRpcUpstream( @@ -92,7 +92,7 @@ open class BitcoinRpcUpstream( override fun isRunning(): Boolean { var runningAny = validatorSubscription != null if (head is Lifecycle) { - runningAny = runningAny || head.isRunning + runningAny = runningAny || head.isRunning() } return runningAny } @@ -100,7 +100,7 @@ open class BitcoinRpcUpstream( override fun start() { log.info("Configured for ${chain.chainName}") if (head is Lifecycle) { - if (!head.isRunning) { + if (!head.isRunning()) { head.start() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt index 4ce4bcc1..07ed5837 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt @@ -5,12 +5,12 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.apache.commons.codec.binary.Hex import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -65,6 +65,6 @@ class BitcoinZMQHead( } override fun isRunning(): Boolean { - return server.isRunning || refreshSubscription != null + return server.isRunning() || refreshSubscription != null } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt index 3e26e5a5..f4df9af9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt @@ -18,11 +18,11 @@ package io.emeraldpay.dshackle.upstream.bitcoin import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Mono import java.time.Duration diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt index 1358f568..fb208deb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ZMQServer.kt @@ -1,7 +1,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin +import io.emeraldpay.dshackle.upstream.Lifecycle import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import org.zeromq.SocketType import org.zeromq.ZContext import org.zeromq.ZMQ diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 259e88a6..fd7f4cf3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -20,20 +20,14 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.ChainFees -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.* +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -85,7 +79,7 @@ open class EthereumMultistream( } override fun isRunning(): Boolean { - return super.isRunning() || reader.isRunning + return super.isRunning() || reader.isRunning() } override fun getReader(): EthereumReader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt index 8501a99b..6f3ea506 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt @@ -30,6 +30,7 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.RekeyingReader import io.emeraldpay.dshackle.reader.RpcReader import io.emeraldpay.dshackle.reader.TransformingReader +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.etherjar.domain.Address @@ -41,7 +42,6 @@ import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import java.util.function.Function /** @@ -174,7 +174,7 @@ open class EthereumReader( override fun isRunning(): Boolean { // TODO should be always running? - return up.isRunning + return true // up.isRunning } override fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index 00637d54..16331153 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -18,11 +18,11 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.BlockValidator +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import org.springframework.scheduling.concurrent.CustomizableThreadFactory import reactor.core.Disposable import reactor.core.publisher.Flux diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 3bbe468f..bb2479df 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -80,7 +80,7 @@ open class EthereumRpcUpstream( } override fun isRunning(): Boolean { - return connector.isRunning + return connector.isRunning() } override fun getApi(): Reader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 1532d5d8..2ed626f6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -17,10 +17,10 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.upstream.BlockValidator +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt index 85ecf596..c9b2b69d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt @@ -2,9 +2,9 @@ package io.emeraldpay.dshackle.upstream.ethereum.connectors import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import org.springframework.context.Lifecycle interface EthereumConnector : Lifecycle { fun getHead(): Head diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index 2b46a499..21fefba1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.MergedHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory @@ -14,7 +15,6 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import java.time.Duration class EthereumRpcConnector( @@ -61,7 +61,7 @@ class EthereumRpcConnector( override fun isRunning(): Boolean { if (head is Lifecycle) { - return head.isRunning + return head.isRunning() } return true } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index 0a248e2c..609e88cf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -36,7 +36,7 @@ class EthereumWsConnector( } override fun isRunning(): Boolean { - return head.isRunning + return head.isRunning() } override fun stop() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index cd872db7..71477c37 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -20,20 +20,14 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.ChainFees -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.* +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -80,7 +74,7 @@ open class EthereumPosMultiStream( } override fun isRunning(): Boolean { - return super.isRunning() || reader.isRunning + return super.isRunning() || reader.isRunning() } override fun getReader(): EthereumReader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt index 50f201c2..78ce2c24 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -31,7 +32,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable open class EthereumPosRpcUpstream( @@ -80,7 +80,7 @@ open class EthereumPosRpcUpstream( } override fun isRunning(): Boolean { - return connector.isRunning + return connector.isRunning() } override fun getApi(): Reader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index a539a21a..35397e43 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -37,7 +38,6 @@ import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index d407d3b5..c0427a06 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -40,7 +41,6 @@ import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 10a005cc..72c03c28 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -40,7 +41,6 @@ import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index ac2db057..78883e1c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -22,12 +22,12 @@ import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -60,7 +60,7 @@ class GrpcHead( * Initiate a new head subscription with connection to the remote */ private fun internalStart(remote: ReactorBlockchainGrpc.ReactorBlockchainStub) { - if (this.isRunning) { + if (this.isRunning()) { stop() } log.debug("Start Head subscription to ${parent.getId()}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index c92fd624..f74cb62c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.AuthConfig 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.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient @@ -69,7 +69,7 @@ class GrpcUpstreams( private val known = HashMap() private val lock = ReentrantLock() - fun start(): Flux { + fun start(): Flux { val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) { NettyChannelBuilder.forAddress(host, port) // 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 } - fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux { + fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux { val current = value.chainsList.filter { Chain.byId(it.chain.number) != Chain.UNSPECIFIED }.mapNotNull { chainDetails -> @@ -138,14 +138,14 @@ class GrpcUpstreams( } val added = current.filter { - it.type == UpstreamChange.ChangeType.ADDED + it.type == UpstreamChangeEvent.ChangeType.ADDED } val removed = known.filterNot { kv -> val stillCurrent = current.any { c -> c.chain == kv.key } stillCurrent }.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) } @@ -172,7 +172,7 @@ class GrpcUpstreams( return sslContext.build() } - fun getOrCreate(chain: Chain): UpstreamChange { + fun getOrCreate(chain: Chain): UpstreamChangeEvent { val metricsTags = listOf( Tag.of("upstream", id), 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 { val current = known[chain] return if (current == null) { @@ -211,14 +211,14 @@ class GrpcUpstreams( created.timeout = this.timeout known[chain] = created created.start() - UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) + UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED) } 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 { val current = known[chain] return if (current == null) { @@ -227,14 +227,14 @@ class GrpcUpstreams( created.timeout = this.timeout known[chain] = created created.start() - UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) + UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED) } 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 { val current = known[chain] return if (current == null) { @@ -243,9 +243,9 @@ class GrpcUpstreams( created.timeout = this.timeout known[chain] = created created.start() - UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED) + UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED) } else { - UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED) + UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED) } } } diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml index 08e5d92d..807a731a 100644 --- a/src/main/resources/log4j2.xml +++ b/src/main/resources/log4j2.xml @@ -41,4 +41,4 @@ - \ No newline at end of file + diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 43352b51..93cc1fbc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -246,9 +246,7 @@ class NativeCallSpec extends Specification { def "Returns error for unsupported chain"() { setup: - def upstreams = Mock(MultistreamHolder) { - _ * it.observeChains() >> Flux.empty() - } + def upstreams = Mock(MultistreamHolder) def nativeCall = nativeCall(upstreams) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 4e6f6eea..f799729d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -4,21 +4,24 @@ import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.quorum.NonEmptyQuorum +import io.emeraldpay.dshackle.upstream.CallTargetsHolder import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.grpc.Chain +import org.springframework.context.ApplicationEventPublisher import spock.lang.Specification class ConfiguredUpstreamsSpec extends Specification { def "Applied quorum to extra methods"() { setup: - def currentUpstreams = Mock(CurrentMultistreamHolder) { - _ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM) - } + def callTargetsHolder = new CallTargetsHolder() def configurer = new ConfiguredUpstreams( - currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig) + Stub(FileResolver), + Stub(UpstreamsConfig), + callTargetsHolder, + Mock(ApplicationEventPublisher) ) def methods = new UpstreamsConfig.Methods( [ @@ -38,11 +41,12 @@ class ConfiguredUpstreamsSpec extends Specification { def "Got static response from extra methods"() { setup: - def currentUpstreams = Mock(CurrentMultistreamHolder) { - _ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM) - } + def callTargetsHolder = new CallTargetsHolder() def configurer = new ConfiguredUpstreams( - currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig) + Stub(FileResolver), + Stub(UpstreamsConfig), + callTargetsHolder, + Mock(ApplicationEventPublisher) ) def methods = new UpstreamsConfig.Methods( [ @@ -61,7 +65,12 @@ class ConfiguredUpstreamsSpec extends Specification { def "Calculate node-id"() { 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: configurer.getHash(node, src) == expected @@ -75,7 +84,12 @@ class ConfiguredUpstreamsSpec extends Specification { def "Calculate node-id conflicting results"() { 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: def h1 = configurer.getHash(null, "hohoho") diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 660b7611..e26cc9f7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -81,20 +81,6 @@ class MultistreamHolderMock implements MultistreamHolder { return upstreams.keySet().toList() } - @Override - Flux observeChains() { - 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 boolean isAvailable(@NotNull Chain chain) { return upstreams.containsKey(chain) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index fbec4d77..7bfab88c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -24,8 +24,10 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.CallTargetsHolder import io.emeraldpay.dshackle.upstream.Multistream 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.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -90,6 +92,21 @@ class TestingCommons { return new CachesFactory(new CacheConfig()) } + static List defaultMultistreams() { + return [ + multistreamWithoutUpstreams(Chain.ETHEREUM), + multistreamClassicWithoutUpstreams(Chain.ETHEREUM_CLASSIC) + ] + } + + static Multistream multistreamWithoutUpstreams(Chain chain) { + return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain)) + } + + static Multistream multistreamClassicWithoutUpstreams(Chain chain) { + return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain)) + } + static FileResolver fileResolver() { return new FileResolver(new File("src/test/resources")) } @@ -126,4 +143,6 @@ class TestingCommons { } static MeterRegistry meterRegistry = new LoggingMeterRegistry() + + static CallTargetsHolder callTargetsHolder = new CallTargetsHolder() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy index 2f38eecd..4e488e9d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy @@ -15,7 +15,7 @@ */ 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.EthereumRpcUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons @@ -26,10 +26,10 @@ class CurrentMultistreamHolderSpec extends Specification { def "add upstream"() { setup: - def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams()) def up = new EthereumPosRpcUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api()) 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: current.getAvailable() == [Chain.ETHEREUM] current.getUpstream(Chain.ETHEREUM).getAll()[0] == up @@ -37,14 +37,15 @@ class CurrentMultistreamHolderSpec extends Specification { def "add multiple upstreams"() { setup: - def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams()) def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up2 = new EthereumRpcUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) when: - current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) - current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) - current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM_CLASSIC).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC, up2, UpstreamChangeEvent.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: current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet() current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up1, up3].toSet() @@ -53,16 +54,16 @@ class CurrentMultistreamHolderSpec extends Specification { def "remove upstream"() { setup: - def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams()) def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up2 = new EthereumRpcUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) def up1_del = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) when: - current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) - current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) - current.update(new UpstreamChange(Chain.ETHEREUM, up3, UpstreamChange.ChangeType.ADDED)) - current.update(new UpstreamChange(Chain.ETHEREUM, up1_del, UpstreamChange.ChangeType.REMOVED)) + current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM_CLASSIC).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC, up2, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up3, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM, up1_del, UpstreamChangeEvent.ChangeType.REMOVED)) then: current.getAvailable().toSet() == [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC].toSet() current.getUpstream(Chain.ETHEREUM).getAll().toSet() == [up3].toSet() @@ -71,7 +72,7 @@ class CurrentMultistreamHolderSpec extends Specification { def "available after adding"() { setup: - def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams()) def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) when: @@ -80,7 +81,7 @@ class CurrentMultistreamHolderSpec extends Specification { !act 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) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy index e1b72360..a6e08be3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy @@ -16,7 +16,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice -import org.springframework.context.Lifecycle +import io.emeraldpay.dshackle.upstream.Lifecycle import reactor.core.publisher.Flux import spock.lang.Specification diff --git a/testing/dshackle/dshackle-basic.yaml b/testing/dshackle/dshackle-basic.yaml index 51318221..56515c78 100644 --- a/testing/dshackle/dshackle-basic.yaml +++ b/testing/dshackle/dshackle-basic.yaml @@ -7,6 +7,7 @@ tls: cluster: upstreams: - id: test-1 + node-id: 1 chain: ethereum methods: enabled: @@ -15,17 +16,21 @@ cluster: options: disable-validation: true connection: - ethereum: - rpc: - url: "http://localhost:18545" + ethereum-pos: + execution: + rpc: + url: "http://localhost:18545" + - id: test-2 + node-id: 2 chain: ethereum options: disable-validation: true connection: - ethereum: - rpc: - url: "http://localhost:18546" + execution: + ethereum-pos: + rpc: + url: "http://localhost:18546" cache: redis: