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

fixing startup

Browse Source
This commit is contained in:
a10zn8
2022-11-16 17:00:07 +04:00
parent eaa68d606b
commit d997d148c4
37 changed files with 154 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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)

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

@@ -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)
}
}

34
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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, EthereumCallSelector>(Chain::class.java)
init {
val casting = mapOf(
companion object {
val casting: Map<BlockchainType, Class<out EthereumLikeMultistream>> = 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()))
}
}
}

17
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
View File

@@ -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)
}
}

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

@@ -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<Byte, Boolean> = 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)
}
}

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

@@ -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<Chain, Multistream>().apply {
multistreams.forEach { this[it.chain] = it }
}
private val chainsBus = Sinks.many()
.multicast()
.directBestEffort<Chain>()
private val updateLock = ReentrantLock()
override fun getUpstream(chain: Chain): Multistream? {
@@ -53,13 +49,6 @@ open class CurrentMultistreamHolder(
.toList()
}
override fun observeChains(): Flux<Chain> {
return Flux.concat(
Flux.fromIterable(getAvailable()),
chainsBus.asFlux()
)
}
override fun isAvailable(chain: Chain): Boolean {
return chainMapping.getValue(chain).isAvailable()
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/Lifecycle.kt Normal file
View File

@@ -0,0 +1,7 @@
package io.emeraldpay.dshackle.upstream
interface Lifecycle {
fun start()
fun stop()
fun isRunning(): Boolean
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt
View File

@@ -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()
}
}

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

@@ -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<Upstream>,
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")
}
}
}

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

@@ -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<Chain>
fun observeChains(): Flux<Chain>
fun isAvailable(chain: Chain): Boolean
}

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

@@ -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<BitcoinUpstream>,
caches: Caches,
callTargetsHolder: CallTargetsHolder
) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty(), callTargetsHolder), Lifecycle {
caches: Caches
) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, 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() {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
View File

@@ -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() {

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

@@ -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

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
View File

@@ -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()
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
View File

@@ -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
}
}

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

@@ -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

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

@@ -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

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

@@ -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<EthereumUpstream>,
caches: Caches,
callTargetsHolder: CallTargetsHolder
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches), callTargetsHolder), EthereumLikeMultistream {
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, 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 {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt
View File

@@ -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() {

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

@@ -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

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

@@ -80,7 +80,7 @@ open class EthereumRpcUpstream(
}
override fun isRunning(): Boolean {
return connector.isRunning
return connector.isRunning()
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {

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

@@ -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

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt
View File

@@ -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

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
View File

@@ -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
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
View File

@@ -36,7 +36,7 @@ class EthereumWsConnector(
}
override fun isRunning(): Boolean {
return head.isRunning
return head.isRunning()
}
override fun stop() {

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

@@ -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<EthereumPosUpstream>,
caches: Caches,
callTargetsHolder: CallTargetsHolder
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches), callTargetsHolder), EthereumLikeMultistream {
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, 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 {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
View File

@@ -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<JsonRpcRequest, JsonRpcResponse> {

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

@@ -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

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

@@ -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

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

@@ -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

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
View File

@@ -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()}")

2
src/main/resources/log4j2.xml
View File

@@ -41,4 +41,4 @@
</Root>
</Loggers>
</Configuration>
</Configuration>

4
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -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()

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

@@ -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<Chain> 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<EthereumPosRpcUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches, TestingCommons.callTargetsHolder)
super(chain, upstreams, caches)
}
EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams) {

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

@@ -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() {

2
src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy
View File

@@ -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

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

@@ -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<Upstream> 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<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches, TestingCommons.callTargetsHolder)
super(chain, upstreams, caches)
}
@Override