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 index d8204b87..3329f93e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -8,11 +8,12 @@ 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 { +open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) { @Bean open fun allMultistreams( cachesFactory: CachesFactory, @@ -22,26 +23,56 @@ open class MultistreamsConfig { .filterNot { it == Chain.UNSPECIFIED } .mapNotNull { chain -> when (BlockchainType.from(chain)) { - BlockchainType.EVM_POS -> EthereumPosMultiStream( - 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 - ) + 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 8fc7e036..b66a125a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -44,12 +44,13 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory +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 @Component @@ -58,15 +59,14 @@ open class ConfiguredUpstreams( 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 -> @@ -111,6 +111,7 @@ open class ConfiguredUpstreams( } upstream?.let { val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED) + log.error("first !!!! Upstream ${event.upstream.getId()} with chain $chain has been added") eventPublisher.publishEvent(event) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index 0402c727..b4865951 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -19,7 +19,6 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.stereotype.Component -import reactor.core.publisher.Flux import reactor.core.publisher.Sinks import java.util.* import java.util.concurrent.ConcurrentHashMap @@ -37,9 +36,6 @@ open class CurrentMultistreamHolder( private val chainMapping = ConcurrentHashMap().apply { multistreams.forEach { this[it.chain] = it } } - private val chainsBus = Sinks.many() - .multicast() - .directBestEffort() private val updateLock = ReentrantLock() override fun getUpstream(chain: Chain): Multistream? { @@ -53,13 +49,6 @@ open class CurrentMultistreamHolder( .toList() } - override fun observeChains(): Flux { - return Flux.concat( - Flux.fromIterable(getAvailable()), - chainsBus.asFlux() - ) - } - override fun isAvailable(chain: Chain): Boolean { return chainMapping.getValue(chain).isAvailable() } 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 80e19481..54fb2dfd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -31,14 +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 @@ -51,8 +52,7 @@ abstract class Multistream( val chain: Chain, private val upstreams: MutableList, val caches: Caches, - val postprocessor: RequestPostprocessor, - val callTargetsHolder: CallTargetsHolder + val postprocessor: RequestPostprocessor ) : Upstream, Lifecycle { companion object { @@ -60,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 @@ -235,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() { @@ -248,6 +251,7 @@ abstract class Multistream( } } lagObserver?.stop() + started = false } fun onHeadUpdated(head: Head) { @@ -321,18 +325,22 @@ abstract class Multistream( } @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.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") + 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) - log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") + if (!started) { + start() + } + log.error("Upstream ${event.upstream.getId()} with chain $chain has been added") } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt index 6e1c4ab0..80b4e564 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt @@ -17,7 +17,6 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.grpc.Chain -import reactor.core.publisher.Flux /** * Holds Multistreams configured for a chain. @@ -25,6 +24,5 @@ import reactor.core.publisher.Flux interface MultistreamHolder { fun getUpstream(chain: Chain): Multistream? fun getAvailable(): List - fun observeChains(): Flux 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 14ebfda0..aca3d567 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -19,22 +19,21 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader 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") open class BitcoinMultistream( chain: Chain, private val sourceUpstreams: MutableList, - caches: Caches, - callTargetsHolder: CallTargetsHolder -) : Multistream(chain, sourceUpstreams as MutableList, caches, RequestPostprocessor.Empty(), callTargetsHolder), Lifecycle { + caches: Caches +) : Multistream(chain, sourceUpstreams as MutableList, caches, RequestPostprocessor.Empty()), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) @@ -136,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 e2fb57f5..fd7f4cf3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -21,13 +21,13 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader 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 @@ -36,9 +36,8 @@ import reactor.core.publisher.Mono open class EthereumMultistream( chain: Chain, val upstreams: MutableList, - caches: Caches, - callTargetsHolder: CallTargetsHolder -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches), callTargetsHolder), EthereumLikeMultistream { + caches: Caches +) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)), EthereumLikeMultistream { companion object { private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) @@ -80,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 3e93a81a..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 @@ -21,13 +21,13 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader 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 @@ -36,9 +36,8 @@ import reactor.core.publisher.Mono open class EthereumPosMultiStream( chain: Chain, val upstreams: MutableList, - caches: Caches, - callTargetsHolder: CallTargetsHolder -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches), callTargetsHolder), EthereumLikeMultistream { + caches: Caches +) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)), EthereumLikeMultistream { companion object { private val log = LoggerFactory.getLogger(EthereumPosMultiStream::class.java) @@ -75,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/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/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 9e34c546..e26cc9f7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -50,7 +50,7 @@ class MultistreamHolderMock implements MultistreamHolder { if (up instanceof EthereumPosMultiStream) { upstreams[chain] = up } else if (up instanceof EthereumPosRpcUpstream) { - upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default(), TestingCommons.callTargetsHolder) + upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default()) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } @@ -59,7 +59,7 @@ class MultistreamHolderMock implements MultistreamHolder { if (up instanceof BitcoinMultistream) { upstreams[chain] = up } else if (up instanceof BitcoinRpcUpstream) { - upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinRpcUpstream], Caches.default(), TestingCommons.callTargetsHolder) + upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinRpcUpstream], Caches.default()) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } @@ -81,11 +81,6 @@ class MultistreamHolderMock implements MultistreamHolder { return upstreams.keySet().toList() } - @Override - Flux observeChains() { - return Flux.fromIterable(getAvailable()) - } - @Override boolean isAvailable(@NotNull Chain chain) { return upstreams.containsKey(chain) @@ -98,7 +93,7 @@ class MultistreamHolderMock implements MultistreamHolder { Head customHead = null EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, TestingCommons.callTargetsHolder) + super(chain, upstreams, caches) } EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index f2c640d7..7bfab88c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -29,7 +29,6 @@ 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.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.BlockHash @@ -80,7 +79,7 @@ class TestingCommons { } static Multistream multistream(EthereumPosRpcUpstreamMock up) { - return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), callTargetsHolder).tap { + return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default()).tap { start() } } @@ -101,11 +100,11 @@ class TestingCommons { } static Multistream multistreamWithoutUpstreams(Chain chain) { - return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), callTargetsHolder) + return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain)) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { - return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), callTargetsHolder) + return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain)) } static FileResolver fileResolver() { 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/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index b432bcb8..8a28f4b3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -50,7 +50,7 @@ class MultistreamSpec extends Specification { setup: 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 aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), TestingCommons.callTargetsHolder) + def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() @@ -206,7 +206,7 @@ class MultistreamSpec extends Specification { def up1 = TestingCommons.upstream("test-1", "internal") def up2 = TestingCommons.upstream("test-2", "external") def up3 = TestingCommons.upstream("test-3", "external") - def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), TestingCommons.callTargetsHolder) + def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default()) expect: multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) @@ -345,7 +345,7 @@ class MultistreamSpec extends Specification { class TestMultistream extends Multistream { TestMultistream(List upstreams, @NotNull RequestPostprocessor postprocessor) { - super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor, TestingCommons.callTargetsHolder) + super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor) } @Override @@ -386,7 +386,7 @@ class MultistreamSpec extends Specification { class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, TestingCommons.callTargetsHolder) + super(chain, upstreams, caches) } @Override