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 b3d2afdd..daa38e6a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -25,14 +25,15 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) @Qualifier("headScheduler") headScheduler: Scheduler, tracer: Tracer, + multistreamEventsScheduler: Scheduler, ): List { return Chain.entries .filterNot { it == Chain.UNSPECIFIED } .map { chain -> if (chain.type == BITCOIN) { - bitcoinMultistream(chain, cachesFactory, headScheduler) + bitcoinMultistream(chain, cachesFactory, headScheduler, multistreamEventsScheduler) } else { - genericMultistream(chain, cachesFactory, headScheduler, tracer) + genericMultistream(chain, cachesFactory, headScheduler, tracer, multistreamEventsScheduler) } } } @@ -42,13 +43,17 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) cachesFactory: CachesFactory, headScheduler: Scheduler, tracer: Tracer, + multistreamEventsScheduler: Scheduler, ): Multistream { val name = "multi-$chain" val cs = ChainSpecificRegistry.resolve(chain) + val caches = cachesFactory.getCaches(chain) return GenericMultistream( chain, + multistreamEventsScheduler, + cs.callSelector(caches), CopyOnWriteArrayList(), - cachesFactory.getCaches(chain), + caches, headScheduler, cs.makeCachingReaderBuilder(tracer), cs::localReaderBuilder, @@ -60,11 +65,13 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) chain: Chain, cachesFactory: CachesFactory, headScheduler: Scheduler, + multistreamEventsScheduler: Scheduler, ): BitcoinMultistream { val name = "multi-bitcoin-$chain" return BitcoinMultistream( chain, + multistreamEventsScheduler, ArrayList(), cachesFactory.getCaches(chain), headScheduler, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt index 4598b093..ac2892e9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt @@ -30,6 +30,11 @@ open class SchedulersConfig { return makeScheduler("head-scheduler", 4, monitoringConfig) } + @Bean + open fun multistreamEventsScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("events-scheduler", 4, monitoringConfig) + } + @Bean open fun wsConnectionResubscribeScheduler(monitoringConfig: MonitoringConfig): Scheduler { return makeScheduler("ws-connection-resubscribe-scheduler", 2, monitoringConfig) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt index 7f32ca15..d6451888 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt @@ -6,12 +6,10 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.startup.ConfiguredUpstreams import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder -import org.springframework.context.ApplicationEventPublisher import org.springframework.stereotype.Component @Component open class ReloadConfigUpstreamService( - private val eventPublisher: ApplicationEventPublisher, private val multistreamHolder: CurrentMultistreamHolder, private val configuredUpstreams: ConfiguredUpstreams, ) { @@ -44,19 +42,19 @@ open class ReloadConfigUpstreamService( chainsToReload.forEach { usedChains.add(it) - multistreamHolder.getUpstream(it) - .getAll() + val ms = multistreamHolder.getUpstream(it) + ms.getAll() .forEach { up -> - eventPublisher.publishEvent(UpstreamChangeEvent(it, up, UpstreamChangeEvent.ChangeType.REMOVED)) + ms.processUpstreamsEvents(UpstreamChangeEvent(it, up, UpstreamChangeEvent.ChangeType.REMOVED)) } } upstreamsToRemove.forEach { pair -> usedChains.add(pair.second) - multistreamHolder.getUpstream(pair.second) - .getAll() + val ms = multistreamHolder.getUpstream(pair.second) + ms.getAll() .find { pair.first == it.getId() } - ?.let { - eventPublisher.publishEvent(UpstreamChangeEvent(pair.second, it, UpstreamChangeEvent.ChangeType.REMOVED)) + ?.let { up -> + ms.processUpstreamsEvents(UpstreamChangeEvent(pair.second, up, UpstreamChangeEvent.ChangeType.REMOVED)) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 29eca620..fb8dde17 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -19,7 +19,6 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.dshackle.BlockchainType.ETHEREUM import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.Global.Companion.nullValue @@ -37,13 +36,11 @@ import io.emeraldpay.dshackle.reader.RpcReader import io.emeraldpay.dshackle.reader.RpcReaderFactory import io.emeraldpay.dshackle.reader.RpcReaderFactory.RpcReaderData import io.emeraldpay.dshackle.reader.SpannedReader -import io.emeraldpay.dshackle.startup.UpstreamChangeEvent 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.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -56,13 +53,11 @@ import org.slf4j.LoggerFactory import org.springframework.cloud.sleuth.Span import org.springframework.cloud.sleuth.Tracer import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth -import org.springframework.context.event.EventListener import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.toMono import reactor.util.context.Context -import java.util.EnumMap @Service open class NativeCall( @@ -78,19 +73,6 @@ open class NativeCall( private val passthrough = config.passthrough var rpcReaderFactory: RpcReaderFactory = RpcReaderFactory.default() - private val ethereumCallSelectors = EnumMap(Chain::class.java) - - @EventListener - fun onUpstreamChangeEvent(event: UpstreamChangeEvent) { - multistreamHolder.getUpstream(event.chain).let { up -> - if (up.chain.type == ETHEREUM) { - ethereumCallSelectors.putIfAbsent( - event.chain, - EthereumCallSelector(up.caches), - ) - } - } - } open fun nativeCall(requestMono: Mono): Flux { return nativeCallResult(requestMono) @@ -305,11 +287,7 @@ open class NativeCall( } // for ethereum the actual block needed for the call may be specified in the call parameters val callSpecificMatcher: Mono = - if (upstream.chain.type == ETHEREUM) { - ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead(), passthrough) - } else { - null - } ?: Mono.empty() + upstream.callSelector?.getMatcher(method, params, upstream.getHead(), passthrough) ?: Mono.empty() return callSpecificMatcher.defaultIfEmpty(Selector.empty).map { csm -> val matcher = Selector.Builder() .withMatcher(csm) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 23036db6..a9ba7751 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -16,97 +16,24 @@ */ package io.emeraldpay.dshackle.startup -import brave.grpc.GrpcTracing -import com.google.common.annotations.VisibleForTesting -import io.emeraldpay.dshackle.BlockchainType.BITCOIN -import io.emeraldpay.dshackle.BlockchainType.ETHEREUM import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.Defaults -import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.config.AuthorizationConfig -import io.emeraldpay.dshackle.config.ChainsConfig -import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig -import io.emeraldpay.dshackle.config.CompressionConfig import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.config.UpstreamsConfig.BitcoinConnection -import io.emeraldpay.dshackle.config.UpstreamsConfig.EthereumPosConnection -import io.emeraldpay.dshackle.config.UpstreamsConfig.HttpEndpoint -import io.emeraldpay.dshackle.config.UpstreamsConfig.RpcConnection import io.emeraldpay.dshackle.foundation.ChainOptions -import io.emeraldpay.dshackle.foundation.ChainOptions.Options -import io.emeraldpay.dshackle.upstream.BlockValidator -import io.emeraldpay.dshackle.upstream.CallTargetsHolder -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.HttpRpcFactory -import io.emeraldpay.dshackle.upstream.MergedHead -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead -import io.emeraldpay.dshackle.upstream.bitcoin.EsploraClient -import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock -import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionFactory -import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory -import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice -import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice -import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice -import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry -import io.emeraldpay.dshackle.upstream.generic.GenericUpstream -import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory -import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD -import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams -import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext -import io.grpc.ClientInterceptor +import io.emeraldpay.dshackle.startup.configure.UpstreamFactory +import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Autowired -import org.springframework.beans.factory.annotation.Qualifier -import org.springframework.beans.factory.annotation.Value import org.springframework.boot.ApplicationArguments import org.springframework.boot.ApplicationRunner -import org.springframework.context.ApplicationEventPublisher import org.springframework.stereotype.Component -import reactor.core.scheduler.Scheduler -import reactor.core.scheduler.Schedulers -import java.net.URI -import java.util.concurrent.Executor -import java.util.concurrent.Executors -import java.util.concurrent.atomic.AtomicInteger -import java.util.function.Function -import kotlin.math.abs @Component open class ConfiguredUpstreams( - private val fileResolver: FileResolver, + private val upstreamFactory: UpstreamFactory, private val config: UpstreamsConfig, - private val compressionConfig: CompressionConfig, - private val callTargets: CallTargetsHolder, - private val eventPublisher: ApplicationEventPublisher, - @Qualifier("grpcChannelExecutor") - private val channelExecutor: Executor, - private val chainsConfig: ChainsConfig, - private val grpcTracing: GrpcTracing, - private val wsConnectionResubscribeScheduler: Scheduler, - @Autowired(required = false) - private val clientSpansInterceptor: ClientInterceptor?, - @Qualifier("headScheduler") - private val headScheduler: Scheduler, - private val wsScheduler: Scheduler, - private val headLivenessScheduler: Scheduler, - private val authorizationConfig: AuthorizationConfig, - private val grpcAuthContext: GrpcAuthContext, + private val multistreamHolder: CurrentMultistreamHolder, ) : ApplicationRunner { - @Value("\${spring.application.max-metadata-size}") - private var maxMetadataSize: Int = Defaults.maxMetadataSize - private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) - private var seq = AtomicInteger(0) - private val hashes: MutableMap = HashMap() - - lateinit var grpcUpstreamsScheduler: Scheduler override fun run(args: ApplicationArguments) { log.debug("Starting upstreams") @@ -121,59 +48,18 @@ open class ConfiguredUpstreams( return@forEach } log.debug("Start upstream ${up.id}") - if (up.connection is UpstreamsConfig.GrpcConnection) { - val options = up.options ?: ChainOptions.PartialOptions() - buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options.buildOptions(), compressionConfig.grpc.clientEnabled) - } else { + if (up.connection !is UpstreamsConfig.GrpcConnection) { val chain = Global.chainById(up.chain) if (chain == Chain.UNSPECIFIED) { log.error("Chain is unknown: ${up.chain}") return@forEach } - val chainConfig = chainsConfig.resolve(up.chain!!) - val options = chainConfig.options - .merge(defaultOptions[chain] ?: ChainOptions.PartialOptions.getDefaults()) - .merge(up.options ?: ChainOptions.PartialOptions()) - .buildOptions() - val upstream = when (chain.type) { - BITCOIN -> { - buildBitcoinUpstream( - up.cast(BitcoinConnection::class.java), - chain, - options, - chainConfig, + val upstream = upstreamFactory.createUpstream(chain.type, up, defaultOptions) + if (upstream != null) { + multistreamHolder.getUpstream(chain) + .processUpstreamsEvents( + UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED), ) - } - - ETHEREUM -> { - val posConn = up.cast(EthereumPosConnection::class.java) - - buildGenericUpstream( - up.nodeId, - up, - posConn.connection?.execution ?: throw IllegalStateException("Empty execution config"), - chain, - options, - chainConfig, - posConn.connection?.upstreamRating ?: 0, - ) - } - - else -> { - buildGenericUpstream( - up.nodeId, - up, - up.connection as RpcConnection, - chain, - options, - chainConfig, - 0, - ) - } - } - upstream?.let { - val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED) - eventPublisher.publishEvent(event) } } } @@ -196,264 +82,4 @@ open class ConfiguredUpstreams( } return defaultOptions } - - fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods { - return if (config.methods != null || config.methodGroups != null) { - ManagedCallMethods( - delegate = callTargets.getDefaultMethods(chain), - enabled = config.methods?.enabled?.map { it.name }?.toSet() ?: emptySet(), - disabled = config.methods?.disabled?.map { it.name }?.toSet() ?: emptySet(), - groupsEnabled = config.methodGroups?.enabled ?: emptySet(), - groupsDisabled = config.methodGroups?.disabled ?: emptySet(), - ).also { - config.methods?.enabled?.forEach { m -> - if (m.quorum != null) { - it.setQuorum(m.name, m.quorum) - } - if (m.static != null) { - it.setStaticResponse(m.name, m.static) - } - } - } - } else { - callTargets.getDefaultMethods(chain) - } - } - - private fun buildGenericUpstream( - nodeId: Int?, - config: UpstreamsConfig.Upstream<*>, - connection: RpcConnection, - chain: Chain, - options: Options, - chainConfig: ChainConfig, - nodeRating: Int, - ): Upstream? { - if (config.connection == null) { - log.warn("Upstream doesn't have connection configuration") - return null - } - - val cs = ChainSpecificRegistry.resolve(chain) - - val connectorFactory = buildConnectorFactory( - config.id!!, - connection, - chain, - NoChoiceWithPriorityForkChoice(nodeRating, config.id!!), - BlockValidator.ALWAYS_VALID, - chainConfig, - ) ?: return null - - val methods = buildMethods(config, chain) - - val hashUrl = connection.let { - if (it.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url - } - val hash = getHash(nodeId, hashUrl!!) - - val upstream = GenericUpstream( - config.id!!, - chain, - hash, - options, - config.role, - methods, - QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)), - chainConfig, - connectorFactory, - eventPublisher, - cs::validator, - cs::labelDetector, - cs::subscriptionTopics, - ) - - upstream.start() - if (!upstream.isRunning) { - log.debug("Upstream ${upstream.getId()} is not running, it can't be added") - return null - } - return upstream - } - - private fun buildBitcoinUpstream( - config: UpstreamsConfig.Upstream, - chain: Chain, - options: Options, - chainConf: ChainConfig, - ): Upstream? { - val conn = config.connection!! - val httpFactory = buildHttpFactory(conn.rpc) - if (httpFactory == null) { - log.warn("Upstream doesn't have API configuration") - return null - } - val directApi = httpFactory.create(config.id, chain) - val esplora = conn.esplora?.let { endpoint -> - val tls = endpoint.tls?.let { tls -> - tls.ca?.let { ca -> - fileResolver.resolve(ca).readBytes() - } - } - EsploraClient(endpoint.url, endpoint.basicAuth, tls) - } - - val extractBlock = ExtractBlock() - val rpcHead = BitcoinRpcHead(directApi, extractBlock, headScheduler = headScheduler) - val head: Head = conn.zeroMq?.let { zeroMq -> - val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock") - val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock, headScheduler) - MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice(), headScheduler) - } ?: rpcHead - - val methods = buildMethods(config, chain) - val upstream = BitcoinRpcUpstream( - config.id - ?: "bitcoin-${seq.getAndIncrement()}", - chain, directApi, head, - options, config.role, - QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)), - methods, esplora, chainConf, - ) - upstream.start() - return upstream - } - - private fun buildGrpcUpstream( - nodeId: Int?, - config: UpstreamsConfig.Upstream, - options: ChainOptions.Options, - compression: Boolean, - ) { - if (!this::grpcUpstreamsScheduler.isInitialized) { - grpcUpstreamsScheduler = Schedulers.fromExecutorService( - Executors.newFixedThreadPool(2), - "GrpcUpstreamsStatuses", - ) - } - val endpoint = config.connection!! - val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}") - val ds = GrpcUpstreams( - config.id!!, - hash, - config.role, - endpoint.host!!, - endpoint.port, - endpoint.auth, - endpoint.tokenAuth, - authorizationConfig, - compression, - fileResolver, - endpoint.upstreamRating, - config.labels, - grpcUpstreamsScheduler, - channelExecutor, - chainsConfig, - grpcTracing, - clientSpansInterceptor, - maxMetadataSize, - headScheduler, - grpcAuthContext, - ).apply { - timeout = options.timeout - } - log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") - ds.start() - .doOnNext { - log.info("Chain ${it.chain} ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}. With caps: ${it.upstream.getCapabilities()}") - } - .subscribe(eventPublisher::publishEvent) - } - - private fun buildHttpFactory(conn: HttpEndpoint?, urls: ArrayList? = null): HttpRpcFactory? { - return conn?.let { endpoint -> - val tls = conn.tls?.let { tls -> - tls.ca?.let { ca -> - fileResolver.resolve(ca).readBytes() - } - } - urls?.add(endpoint.url) - HttpRpcFactory(endpoint.url.toString(), conn.basicAuth, tls) - } - } - - private fun buildWsFactory( - id: String, - chain: Chain, - conn: RpcConnection, - urls: ArrayList? = null, - ): WsConnectionPoolFactory? { - return conn.ws?.let { endpoint -> - val wsConnectionFactory = WsConnectionFactory( - id, - chain, - endpoint.url, - endpoint.origin ?: URI("http://localhost"), - wsScheduler, - ).apply { - config = endpoint - basicAuth = endpoint.basicAuth - } - val wsApi = WsConnectionPoolFactory( - id, - endpoint.connections, - wsConnectionFactory, - ) - urls?.add(endpoint.url) - wsApi - } - } - - private fun buildConnectorFactory( - id: String, - conn: RpcConnection, - chain: Chain, - forkChoice: ForkChoice, - blockValidator: BlockValidator, - chainsConf: ChainConfig, - ): GenericConnectorFactory? { - val urls = ArrayList() - val wsFactoryApi = buildWsFactory(id, chain, conn, urls) - val httpFactory = buildHttpFactory(conn.rpc, urls) - log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val connectorFactory = - GenericConnectorFactory( - conn.resolveMode(), - wsFactoryApi, - httpFactory, - forkChoice, - blockValidator, - wsConnectionResubscribeScheduler, - headScheduler, - headLivenessScheduler, - chainsConf.expectedBlockTime, - ) - if (!connectorFactory.isValid()) { - log.warn("Upstream configuration is invalid (probably no http endpoint)") - return null - } - return connectorFactory - } - - @VisibleForTesting - private fun getHash(nodeId: Int?, obj: Any): Byte = - nodeId?.toByte() ?: (obj.hashCode() % 255).let { - if (it == 0) 1 else it - }.let { nonZeroHash -> - listOf>( - Function { i -> i }, - Function { i -> (-i) }, - Function { i -> 127 - abs(i) }, - Function { i -> abs(i) - 128 }, - ).map { - it.apply(nonZeroHash).toByte() - }.firstOrNull { - hashes[it] != true - }?.let { - hashes[it] = true - it - } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first { - it != 0 && hashes[it.toByte()] != true - }.toByte() - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt new file mode 100644 index 00000000..c6c5ccc3 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt @@ -0,0 +1,77 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.FileResolver +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.foundation.ChainOptions +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.CallTargetsHolder +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead +import io.emeraldpay.dshackle.upstream.bitcoin.EsploraClient +import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock +import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import org.springframework.stereotype.Component +import reactor.core.scheduler.Scheduler +import java.util.concurrent.atomic.AtomicInteger + +@Component +class BitcoinUpstreamCreator( + chainsConfig: ChainsConfig, + callTargets: CallTargetsHolder, + private val genericConnectorFactoryCreator: ConnectorFactoryCreator, + private val fileResolver: FileResolver, + private val headScheduler: Scheduler, +) : UpstreamCreator(chainsConfig, callTargets) { + private var seq = AtomicInteger(0) + + override fun createUpstream( + upstreamsConfig: UpstreamsConfig.Upstream<*>, + chain: Chain, + options: ChainOptions.Options, + chainConf: ChainsConfig.ChainConfig, + ): Upstream? { + val config = upstreamsConfig.cast(UpstreamsConfig.BitcoinConnection::class.java) + val conn = config.connection!! + val httpFactory = genericConnectorFactoryCreator.buildHttpFactory(conn.rpc) + if (httpFactory == null) { + log.warn("Upstream doesn't have API configuration") + return null + } + val directApi = httpFactory.create(config.id, chain) + val esplora = conn.esplora?.let { endpoint -> + val tls = endpoint.tls?.let { tls -> + tls.ca?.let { ca -> + fileResolver.resolve(ca).readBytes() + } + } + EsploraClient(endpoint.url, endpoint.basicAuth, tls) + } + + val extractBlock = ExtractBlock() + val rpcHead = BitcoinRpcHead(directApi, extractBlock, headScheduler = headScheduler) + val head: Head = conn.zeroMq?.let { zeroMq -> + val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock") + val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock, headScheduler) + MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice(), headScheduler) + } ?: rpcHead + + val methods = buildMethods(config, chain) + val upstream = BitcoinRpcUpstream( + config.id + ?: "bitcoin-${seq.getAndIncrement()}", + chain, directApi, head, + options, config.role, + QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)), + methods, esplora, chainConf, + ) + upstream.start() + return upstream + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/ConnectorFactoryCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/ConnectorFactoryCreator.kt new file mode 100644 index 00000000..873e10dd --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/ConnectorFactoryCreator.kt @@ -0,0 +1,23 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.upstream.BlockValidator +import io.emeraldpay.dshackle.upstream.HttpRpcFactory +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory +import java.net.URI + +interface ConnectorFactoryCreator { + fun createConnectorFactoryCreator( + id: String, + conn: UpstreamsConfig.RpcConnection, + chain: Chain, + forkChoice: ForkChoice, + blockValidator: BlockValidator, + chainsConf: ChainsConfig.ChainConfig, + ): ConnectorFactory? + + fun buildHttpFactory(conn: UpstreamsConfig.HttpEndpoint?, urls: ArrayList? = null): HttpRpcFactory? +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt new file mode 100644 index 00000000..3f653db8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt @@ -0,0 +1,36 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.foundation.ChainOptions +import io.emeraldpay.dshackle.upstream.CallTargetsHolder +import io.emeraldpay.dshackle.upstream.Upstream +import org.springframework.stereotype.Component + +@Component +class EthereumUpstreamCreator( + chainsConfig: ChainsConfig, + callTargets: CallTargetsHolder, + genericConnectorFactoryCreator: ConnectorFactoryCreator, +) : GenericUpstreamCreator(chainsConfig, callTargets, genericConnectorFactoryCreator) { + + override fun createUpstream( + upstreamsConfig: UpstreamsConfig.Upstream<*>, + chain: Chain, + options: ChainOptions.Options, + chainConf: ChainsConfig.ChainConfig, + ): Upstream? { + val posConn = upstreamsConfig.cast(UpstreamsConfig.EthereumPosConnection::class.java) + + return buildGenericUpstream( + upstreamsConfig.nodeId, + upstreamsConfig, + posConn.connection?.execution ?: throw IllegalStateException("Empty execution config"), + chain, + options, + chainConf, + posConn.connection?.upstreamRating ?: 0, + ) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt new file mode 100644 index 00000000..2b68dcee --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt @@ -0,0 +1,98 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.FileResolver +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.upstream.BlockValidator +import io.emeraldpay.dshackle.upstream.HttpRpcFactory +import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionFactory +import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory +import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory +import org.slf4j.LoggerFactory +import org.springframework.stereotype.Component +import reactor.core.scheduler.Scheduler +import java.net.URI + +@Component +class GenericConnectorFactoryCreator( + private val fileResolver: FileResolver, + private val wsConnectionResubscribeScheduler: Scheduler, + private val headScheduler: Scheduler, + private val wsScheduler: Scheduler, + private val headLivenessScheduler: Scheduler, +) : ConnectorFactoryCreator { + private val log = LoggerFactory.getLogger(this::class.java) + + override fun createConnectorFactoryCreator( + id: String, + conn: UpstreamsConfig.RpcConnection, + chain: Chain, + forkChoice: ForkChoice, + blockValidator: BlockValidator, + chainsConf: ChainsConfig.ChainConfig, + ): ConnectorFactory? { + val urls = ArrayList() + val wsFactoryApi = buildWsFactory(id, chain, conn, urls) + val httpFactory = buildHttpFactory(conn.rpc, urls) + log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") + val connectorFactory = + GenericConnectorFactory( + conn.resolveMode(), + wsFactoryApi, + httpFactory, + forkChoice, + blockValidator, + wsConnectionResubscribeScheduler, + headScheduler, + headLivenessScheduler, + chainsConf.expectedBlockTime, + ) + if (!connectorFactory.isValid()) { + log.warn("Upstream configuration is invalid (probably no http endpoint)") + return null + } + return connectorFactory + } + + override fun buildHttpFactory(conn: UpstreamsConfig.HttpEndpoint?, urls: ArrayList?): HttpRpcFactory? { + return conn?.let { endpoint -> + val tls = conn.tls?.let { tls -> + tls.ca?.let { ca -> + fileResolver.resolve(ca).readBytes() + } + } + urls?.add(endpoint.url) + HttpRpcFactory(endpoint.url.toString(), conn.basicAuth, tls) + } + } + + private fun buildWsFactory( + id: String, + chain: Chain, + conn: UpstreamsConfig.RpcConnection, + urls: ArrayList? = null, + ): WsConnectionPoolFactory? { + return conn.ws?.let { endpoint -> + val wsConnectionFactory = WsConnectionFactory( + id, + chain, + endpoint.url, + endpoint.origin ?: URI("http://localhost"), + wsScheduler, + ).apply { + config = endpoint + basicAuth = endpoint.basicAuth + } + val wsApi = WsConnectionPoolFactory( + id, + endpoint.connections, + wsConnectionFactory, + ) + urls?.add(endpoint.url) + wsApi + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt new file mode 100644 index 00000000..bb0c0376 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt @@ -0,0 +1,119 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.foundation.ChainOptions +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.BlockValidator +import io.emeraldpay.dshackle.upstream.CallTargetsHolder +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice +import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry +import io.emeraldpay.dshackle.upstream.generic.GenericUpstream +import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory +import org.springframework.stereotype.Component +import java.util.function.Function +import kotlin.math.abs + +@Component +open class GenericUpstreamCreator( + chainsConfig: ChainsConfig, + callTargets: CallTargetsHolder, + private val genericConnectorFactoryCreator: ConnectorFactoryCreator, +) : UpstreamCreator(chainsConfig, callTargets) { + private val hashes: MutableMap = HashMap() + + override fun createUpstream( + upstreamsConfig: UpstreamsConfig.Upstream<*>, + chain: Chain, + options: ChainOptions.Options, + chainConf: ChainsConfig.ChainConfig, + ): Upstream? { + return buildGenericUpstream( + upstreamsConfig.nodeId, + upstreamsConfig, + upstreamsConfig.connection as UpstreamsConfig.RpcConnection, + chain, + options, + chainConf, + 0, + ) + } + + protected fun buildGenericUpstream( + nodeId: Int?, + config: UpstreamsConfig.Upstream<*>, + connection: UpstreamsConfig.RpcConnection, + chain: Chain, + options: ChainOptions.Options, + chainConfig: ChainsConfig.ChainConfig, + nodeRating: Int, + ): Upstream? { + if (config.connection == null) { + log.warn("Upstream doesn't have connection configuration") + return null + } + + val cs = ChainSpecificRegistry.resolve(chain) + + val connectorFactory = genericConnectorFactoryCreator.createConnectorFactoryCreator( + config.id!!, + connection, + chain, + NoChoiceWithPriorityForkChoice(nodeRating, config.id!!), + BlockValidator.ALWAYS_VALID, + chainConfig, + ) ?: return null + + val methods = buildMethods(config, chain) + + val hashUrl = connection.let { + if (it.connectorMode == GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url + } + val hash = getHash(nodeId, hashUrl!!) + + val upstream = GenericUpstream( + config.id!!, + chain, + hash, + options, + config.role, + methods, + QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)), + chainConfig, + connectorFactory, + cs::validator, + cs::labelDetector, + cs::subscriptionTopics, + ) + + upstream.start() + if (!upstream.isRunning) { + log.debug("Upstream ${upstream.getId()} is not running, it can't be added") + return null + } + return upstream + } + + private fun getHash(nodeId: Int?, obj: Any): Byte = + nodeId?.toByte() ?: (obj.hashCode() % 255).let { + if (it == 0) 1 else it + }.let { nonZeroHash -> + listOf>( + Function { i -> i }, + Function { i -> (-i) }, + Function { i -> 127 - abs(i) }, + Function { i -> abs(i) - 128 }, + ).map { + it.apply(nonZeroHash).toByte() + }.firstOrNull { + hashes[it] != true + }?.let { + hashes[it] = true + it + } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first { + it != 0 && hashes[it.toByte()] != true + }.toByte() + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt new file mode 100644 index 00000000..8eba6746 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt @@ -0,0 +1,67 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.foundation.ChainOptions +import io.emeraldpay.dshackle.upstream.CallTargetsHolder +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods +import org.slf4j.Logger +import org.slf4j.LoggerFactory + +abstract class UpstreamCreator( + private val chainsConfig: ChainsConfig, + private val callTargets: CallTargetsHolder, +) { + protected val log: Logger = LoggerFactory.getLogger(this::class.java) + + fun createUpstream( + upstreamsConfig: UpstreamsConfig.Upstream<*>, + defaultOptions: Map, + ): Upstream? { + val chain = Global.chainById(upstreamsConfig.chain) + if (chain == Chain.UNSPECIFIED) { + throw IllegalArgumentException("Chain is unknown: ${upstreamsConfig.chain}") + } + val chainConfig = chainsConfig.resolve(upstreamsConfig.chain!!) + val options = chainConfig.options + .merge(defaultOptions[chain] ?: ChainOptions.PartialOptions.getDefaults()) + .merge(upstreamsConfig.options ?: ChainOptions.PartialOptions()) + .buildOptions() + + return createUpstream(upstreamsConfig, chain, options, chainConfig) + } + + protected abstract fun createUpstream( + upstreamsConfig: UpstreamsConfig.Upstream<*>, + chain: Chain, + options: ChainOptions.Options, + chainConf: ChainsConfig.ChainConfig, + ): Upstream? + + protected fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods { + return if (config.methods != null || config.methodGroups != null) { + ManagedCallMethods( + delegate = callTargets.getDefaultMethods(chain), + enabled = config.methods?.enabled?.map { it.name }?.toSet() ?: emptySet(), + disabled = config.methods?.disabled?.map { it.name }?.toSet() ?: emptySet(), + groupsEnabled = config.methodGroups?.enabled ?: emptySet(), + groupsDisabled = config.methodGroups?.disabled ?: emptySet(), + ).also { + config.methods?.enabled?.forEach { m -> + if (m.quorum != null) { + it.setQuorum(m.name, m.quorum) + } + if (m.static != null) { + it.setStaticResponse(m.name, m.static) + } + } + } + } else { + callTargets.getDefaultMethods(chain) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt new file mode 100644 index 00000000..c7097156 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt @@ -0,0 +1,28 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.BlockchainType +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.foundation.ChainOptions +import io.emeraldpay.dshackle.upstream.Upstream +import org.springframework.stereotype.Component + +@Component +class UpstreamFactory( + private val genericUpstreamCreator: GenericUpstreamCreator, + private val ethereumUpstreamCreator: EthereumUpstreamCreator, + private val bitcoinUpstreamCreator: BitcoinUpstreamCreator, +) { + + fun createUpstream( + type: BlockchainType, + upstreamsConfig: UpstreamsConfig.Upstream<*>, + defaultOptions: Map, + ): Upstream? { + return when (type) { + BlockchainType.ETHEREUM -> ethereumUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions) + BlockchainType.BITCOIN -> bitcoinUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions) + else -> genericUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index bc896c56..44b273d2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.config.ChainsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.CallMethods import org.slf4j.LoggerFactory import reactor.core.publisher.Flux @@ -56,7 +57,7 @@ abstract class DefaultUpstream( private val statusStream = Sinks.many() .multicast() .directBestEffort() - protected val stateStream: Sinks.Many = Sinks.many() + protected val stateEventStream: Sinks.Many = Sinks.many() .multicast() .directBestEffort() @@ -109,8 +110,8 @@ abstract class DefaultUpstream( return statusStream.asFlux().distinctUntilChanged() } - override fun observeState(): Flux { - return stateStream.asFlux() + override fun observeState(): Flux { + return stateEventStream.asFlux() } override fun setLag(lag: Long) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 7980918b..72a9dc01 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels 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.calls.CallSelector import io.micrometer.core.instrument.Gauge import io.micrometer.core.instrument.Meter import io.micrometer.core.instrument.Metrics @@ -34,17 +35,13 @@ 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.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 reactor.core.publisher.Sinks +import reactor.core.scheduler.Scheduler import java.time.Duration import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock /** * Aggregation of multiple upstreams responding to a single blockchain @@ -52,8 +49,9 @@ import kotlin.concurrent.withLock abstract class Multistream( val chain: Chain, val caches: Caches, + val callSelector: CallSelector?, + multistreamEventsScheduler: Scheduler, ) : Upstream, Lifecycle { - abstract fun getUpstreams(): MutableList abstract fun addUpstreamInternal(u: Upstream) @@ -66,8 +64,6 @@ abstract class Multistream( private var started = false private var cacheSubscription: Disposable? = null - private val reconfigLock = ReentrantLock() - private val eventLock = ReentrantLock() @Volatile private var callMethods: CallMethods? = null @@ -94,6 +90,9 @@ abstract class Multistream( private val updateUpstreams = Sinks.many() .multicast() .directBestEffort() + private val upstreamsSink = Sinks.many() + .multicast() + .directBestEffort() override fun getSubscriptionTopics(): List { return getEgressSubscription().getAvailableTopics() @@ -127,10 +126,6 @@ abstract class Multistream( ) } - open fun init() { - onUpstreamsUpdated() - } - init { UpstreamAvailability.entries.forEach { status -> Metrics.gauge( @@ -149,6 +144,12 @@ abstract class Multistream( ) { getAll().size.toDouble() } + + upstreamsSink.asFlux() + .publishOn(multistreamEventsScheduler) + .subscribe { + onUpstreamChange(it) + } } /** @@ -206,36 +207,34 @@ abstract class Multistream( throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream") } - open fun onUpstreamsUpdated() { - reconfigLock.withLock { - val upstreams = getAll() - val availableUpstreams = upstreams.filter { it.isAvailable() } - availableUpstreams.map { it.getMethods() }.let { - callMethods = AggregatedCallMethods(it) - } - capabilities = if (upstreams.isEmpty()) { - emptySet() - } else { - availableUpstreams.map { up -> - up.getCapabilities() - }.let { - if (it.isNotEmpty()) { - it.reduce { acc, curr -> acc + curr } - } else { - emptySet() - } + protected open fun onUpstreamsUpdated() { + val upstreams = getAll() + val availableUpstreams = upstreams.filter { it.isAvailable() } + availableUpstreams.map { it.getMethods() }.let { + callMethods = AggregatedCallMethods(it) + } + capabilities = if (upstreams.isEmpty()) { + emptySet() + } else { + availableUpstreams.map { up -> + up.getCapabilities() + }.let { + if (it.isNotEmpty()) { + it.reduce { acc, curr -> acc + curr } + } else { + emptySet() } } - quorumLabels = getQuorumLabels(availableUpstreams) - when { - upstreams.size == 1 -> { - lagObserver?.stop() - lagObserver = null - upstreams[0].setLag(0) - } + } + quorumLabels = getQuorumLabels(availableUpstreams) + when { + upstreams.size == 1 -> { + lagObserver?.stop() + lagObserver = null + upstreams[0].setLag(0) + } - upstreams.size > 1 -> if (lagObserver == null) lagObserver = makeLagObserver() - } + upstreams.size > 1 -> if (lagObserver == null) lagObserver = makeLagObserver() } } @@ -265,7 +264,7 @@ abstract class Multistream( ).distinct() } - override fun observeState(): Flux { + override fun observeState(): Flux { return Flux.empty() } @@ -319,8 +318,9 @@ abstract class Multistream( .distinctUntilChanged { it.getId() }.flatMap { upstream -> - val statusStream = upstream.observeStatus().map { upstream } - val stateStream = upstream.observeState().map { upstream } + val statusStream = upstream.observeStatus() + .map { UpstreamChangeEvent(this.chain, upstream, UpstreamChangeEvent.ChangeType.UPDATED) } + val stateStream = upstream.observeState() Flux.merge(stateStream, statusStream) .takeUntilOther( subscribeRemovedUpstreams() @@ -330,7 +330,7 @@ abstract class Multistream( ) } .subscribe { - onUpstreamChange(UpstreamChangeEvent(this.chain, it, UpstreamChangeEvent.ChangeType.UPDATED)) + this.processUpstreamsEvents(it) } } @@ -350,11 +350,9 @@ abstract class Multistream( } fun onHeadUpdated(head: Head) { - reconfigLock.withLock { - cacheSubscription?.dispose() - cacheSubscription = head.getFlux().subscribe { - caches.cache(Caches.Tag.LATEST, it) - } + cacheSubscription?.dispose() + cacheSubscription = head.getFlux().subscribe { + caches.cache(Caches.Tag.LATEST, it) } caches.setHead(head) } @@ -418,47 +416,49 @@ abstract class Multistream( return event.chain == this.chain } - @EventListener - @Order(Ordered.HIGHEST_PRECEDENCE) - fun onUpstreamChange(event: UpstreamChangeEvent) { + fun processUpstreamsEvents(event: UpstreamChangeEvent) { + upstreamsSink.emitNext( + event, + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + } + + private fun onUpstreamChange(event: UpstreamChangeEvent) { val chain = event.chain if (this.chain == chain) { - eventLock.withLock { - log.debug("Processing event $event") - when (event.type) { - UpstreamChangeEvent.ChangeType.REVALIDATED -> {} - UpstreamChangeEvent.ChangeType.UPDATED -> { - onUpstreamsUpdated() - updateUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - } + log.debug("Processing event {}", event) + when (event.type) { + UpstreamChangeEvent.ChangeType.REVALIDATED -> {} + UpstreamChangeEvent.ChangeType.UPDATED -> { + onUpstreamsUpdated() + updateUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + } - UpstreamChangeEvent.ChangeType.ADDED -> { - if (!started) { - start() - } - if (event.upstream is CachesEnabled) { - event.upstream.setCaches(caches) - } - addUpstream(event.upstream).takeIf { it }?.let { - try { - addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - onUpstreamsUpdated() - log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") - } catch (e: Sinks.EmissionException) { - log.error("error during event processing $event", e) - } + UpstreamChangeEvent.ChangeType.ADDED -> { + if (!started) { + start() + } + if (event.upstream is CachesEnabled) { + event.upstream.setCaches(caches) + } + addUpstream(event.upstream).takeIf { it }?.let { + try { + addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + onUpstreamsUpdated() + log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") + } catch (e: Sinks.EmissionException) { + log.error("error during event processing $event", e) } } + } - UpstreamChangeEvent.ChangeType.REMOVED -> { - removeUpstream(event.upstream.getId()).takeIf { it }?.let { - try { - removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - onUpstreamsUpdated() - log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") - } catch (e: Sinks.EmissionException) { - log.error("error during event processing $event", e) - } + UpstreamChangeEvent.ChangeType.REMOVED -> { + removeUpstream(event.upstream.getId()).takeIf { it }?.let { + try { + removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + onUpstreamsUpdated() + log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") + } catch (e: Sinks.EmissionException) { + log.error("error during event processing $event", e) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 422f423d..4976303c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.reader.JsonRpcReader +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.CallMethods import reactor.core.publisher.Flux @@ -26,7 +27,7 @@ interface Upstream : Lifecycle { fun isAvailable(): Boolean fun getStatus(): UpstreamAvailability fun observeStatus(): Flux - fun observeState(): Flux + fun observeState(): Flux fun getHead(): Head /** 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 03f2bcdb..64d5fa6d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -40,10 +40,11 @@ import reactor.core.scheduler.Scheduler @Suppress("UNCHECKED_CAST") open class BitcoinMultistream( chain: Chain, + multistreamEventsScheduler: Scheduler, private val sourceUpstreams: MutableList, caches: Caches, private val headScheduler: Scheduler, -) : Multistream(chain, caches), Lifecycle { +) : Multistream(chain, caches, null, multistreamEventsScheduler), Lifecycle { private var head: Head = EmptyHead() private var esplora = sourceUpstreams.find { it.esploraClient != null }?.esploraClient @@ -59,13 +60,6 @@ open class BitcoinMultistream( sourceUpstreams.add(u as BitcoinUpstream) } - override fun init() { - if (sourceUpstreams.size > 0) { - head = updateHead() - } - super.init() - } - open fun getXpubAddresses(): XpubAddresses? { return xpubAddresses } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallSelector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallSelector.kt new file mode 100644 index 00000000..5a6025f0 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallSelector.kt @@ -0,0 +1,10 @@ +package io.emeraldpay.dshackle.upstream.calls + +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Selector +import reactor.core.publisher.Mono + +interface CallSelector { + + fun getMatcher(method: String, params: String, head: Head, passthrough: Boolean): Mono +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt index b1eae08e..78daa36a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt @@ -32,7 +32,7 @@ import java.util.Objects */ class EthereumCallSelector( private val caches: Caches, -) { +) : CallSelector { companion object { private val log = LoggerFactory.getLogger(EthereumCallSelector::class.java) @@ -68,7 +68,7 @@ class EthereumCallSelector( * @param method JSON RPC name * @param params JSON-encoded list of parameters for the method */ - fun getMatcher(method: String, params: String, head: Head, passthrough: Boolean): Mono { + override fun getMatcher(method: String, params: String, head: Head, passthrough: Boolean): Mono { if (method in DefaultEthereumMethods.withFilterIdMethods) { return sameUpstreamMatcher(params) } else if (!passthrough) { // passthrough indicates we should match only labels diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt index 2f9bc3de..105874ed 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -1,6 +1,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.foundation.ChainOptions.Options @@ -15,6 +16,8 @@ import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.CallSelector +import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription @@ -99,4 +102,8 @@ object EthereumChainSpecific : ChainSpecific { override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { return EthereumWsIngressSubscription(ws) } + + override fun callSelector(caches: Caches): CallSelector { + return EthereumCallSelector(caches) + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index 61b8b835..00abe582 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions import io.emeraldpay.dshackle.upstream.polkadot.PolkadotChainSpecific @@ -58,6 +59,8 @@ interface ChainSpecific { fun subscriptionTopics(upstream: GenericUpstream): List fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription + + fun callSelector(caches: Caches): CallSelector? } object ChainSpecificRegistry { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt index 7fb8f013..31173f9c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt @@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Selector.Matcher import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream import org.springframework.util.ConcurrentReferenceHashMap @@ -44,13 +45,15 @@ import reactor.core.scheduler.Scheduler open class GenericMultistream( chain: Chain, + multistreamEventsScheduler: Scheduler, + callSelector: CallSelector?, private val upstreams: MutableList, caches: Caches, private val headScheduler: Scheduler, cachingReaderBuilder: CachingReaderBuilder, private val localReaderBuilder: LocalReaderBuilder, private val subscriptionBuilder: SubscriptionBuilder, -) : Multistream(chain, caches) { +) : Multistream(chain, caches, callSelector, multistreamEventsScheduler) { private val cachingReader = cachingReaderBuilder(this, caches, getMethodsFactory()) @@ -68,19 +71,8 @@ open class GenericMultistream( headScheduler, ) - init { - this.init() - } - private var subscription: EgressSubscription = subscriptionBuilder(this) - override fun init() { - if (upstreams.size > 0) { - upstreams.forEach { addHead(it) } - } - super.init() - } - private val filteredHeads: MutableMap = ConcurrentReferenceHashMap(16, WEAK) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index 51668f4f..97e5c0cd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt @@ -21,7 +21,6 @@ import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector -import org.springframework.context.ApplicationEventPublisher import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -39,7 +38,6 @@ open class GenericUpstream( private val node: QuorumForLabels.QuorumItem?, chainConfig: ChainsConfig.ChainConfig, connectorFactory: ConnectorFactory, - private val eventPublisher: ApplicationEventPublisher?, validatorBuilder: UpstreamValidatorBuilder, labelsDetectorBuilder: LabelsDetectorBuilder, private val subscriptionTopics: (GenericUpstream) -> List, @@ -132,13 +130,9 @@ open class GenericUpstream( ValidateUpstreamSettingsResult.UPSTREAM_VALID -> { upstreamStart() - eventPublisher?.publishEvent( - UpstreamChangeEvent( - chain, - this, - UpstreamChangeEvent.ChangeType.ADDED, - ), - ) + stateEventStream.emitNext( + UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.ADDED), + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } disposeValidationSettingsSubscription() } @@ -166,7 +160,9 @@ open class GenericUpstream( } livenessSubscription = connector.hasLiveSubscriptionHead().subscribe({ hasLiveSubscriptionHead.set(it) - stateStream.emitNext(true) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + stateEventStream.emitNext( + UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED), + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } }, { log.debug("Error while checking live subscription for ${getId()}", it) },) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt index 17ef9419..3cfac622 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt @@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties import com.fasterxml.jackson.annotation.JsonProperty import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId @@ -19,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.NoopCachingReader import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder import io.emeraldpay.dshackle.upstream.generic.ChainSpecific @@ -104,6 +106,10 @@ object PolkadotChainSpecific : ChainSpecific { override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { return PolkadotIngressSubscription(ws) } + + override fun callSelector(caches: Caches): CallSelector? { + return null + } } @JsonIgnoreProperties(ignoreUnknown = true) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt index abbd07a3..d0a9020b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt @@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties import com.fasterxml.jackson.annotation.JsonProperty import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId @@ -21,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.NoopCachingReader import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder import io.emeraldpay.dshackle.upstream.generic.ChainSpecific @@ -102,6 +104,10 @@ object StarknetChainSpecific : ChainSpecific { override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { return NoIngressSubscription() } + + override fun callSelector(caches: Caches): CallSelector? { + return null + } } @JsonIgnoreProperties(ignoreUnknown = true) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 8ff0659c..06fb63bb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -28,11 +28,13 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.RpcReader import io.emeraldpay.dshackle.reader.RpcReaderFactory +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods +import io.emeraldpay.dshackle.upstream.generic.GenericUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -601,9 +603,10 @@ class NativeCallSpec extends Specification { new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), [] as Set, [] as Set, ["filter"] as Set, [] as Set ) - def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, TestingCommons.upstream( - TestingCommons.api(), methods - )) + def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, new ArrayList()) + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, TestingCommons.upstream(TestingCommons.api(), methods), UpstreamChangeEvent.ChangeType.ADDED) + ) multistream.customHead = Mock(Head) def multistreamHolder = Mock(MultistreamHolder) { _ * it.observeChains() >> Flux.empty() @@ -636,9 +639,10 @@ class NativeCallSpec extends Specification { new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), [] as Set, [] as Set, ["filter"] as Set, [] as Set ) - def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, TestingCommons.upstream( - TestingCommons.api(), methods - )) + def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, new ArrayList()) + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, TestingCommons.upstream(TestingCommons.api(), methods), UpstreamChangeEvent.ChangeType.ADDED) + ) multistream.customHead = Mock(Head) def multistreamHolder = Mock(MultistreamHolder) { _ * it.observeChains() >> Flux.empty() diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy deleted file mode 100644 index 37778a79..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ /dev/null @@ -1,191 +0,0 @@ -package io.emeraldpay.dshackle.startup - -import brave.Tracing -import brave.grpc.GrpcTracing -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.FileResolver -import io.emeraldpay.dshackle.config.AuthorizationConfig -import io.emeraldpay.dshackle.config.ChainsConfig -import io.emeraldpay.dshackle.config.CompressionConfig -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.quorum.NotNullQuorum -import io.emeraldpay.dshackle.upstream.CallTargetsHolder -import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods -import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext -import org.springframework.context.ApplicationEventPublisher -import reactor.core.scheduler.Schedulers -import spock.lang.Specification - -import java.util.concurrent.Executors - -class ConfiguredUpstreamsSpec extends Specification { - - def "Applied quorum to extra methods"() { - setup: - def callTargetsHolder = new CallTargetsHolder() - def configurer = new ConfiguredUpstreams( - Stub(FileResolver), - new UpstreamsConfig(), - Stub(CompressionConfig), - callTargetsHolder, - Mock(ApplicationEventPublisher), - Executors.newFixedThreadPool(1), - ChainsConfig.default(), - GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), - null, - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - AuthorizationConfig.default(), - new GrpcAuthContext() - ) - def methods = new UpstreamsConfig.Methods( - [ - new UpstreamsConfig.Method("foo_bar", null, null), - new UpstreamsConfig.Method("foo_bar", "not_empty", null) - ] as Set, - [] as Set - ) - def upstream = new UpstreamsConfig.Upstream() - upstream.methods = methods - when: - def act = configurer.buildMethods(upstream, Chain.ETHEREUM__MAINNET) - then: - act instanceof ManagedCallMethods - act.createQuorumFor("foo_bar") instanceof NotNullQuorum - } - - def "Got static response from extra methods"() { - setup: - def callTargetsHolder = new CallTargetsHolder() - def configurer = new ConfiguredUpstreams( - Stub(FileResolver), - new UpstreamsConfig(), - Stub(CompressionConfig), - callTargetsHolder, - Mock(ApplicationEventPublisher), - Executors.newFixedThreadPool(1), - ChainsConfig.default(), - GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), - null, - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - AuthorizationConfig.default(), - new GrpcAuthContext() - ) - def methods = new UpstreamsConfig.Methods( - [ - new UpstreamsConfig.Method("foo_bar", null, "static_response") - ] as Set, - [] as Set - ) - def upstream = new UpstreamsConfig.Upstream() - upstream.methods = methods - when: - def act = configurer.buildMethods(upstream, Chain.ETHEREUM__MAINNET) - then: - act instanceof ManagedCallMethods - new String(act.executeHardcoded("foo_bar")) == "\"static_response\"" - } - - def "Calculate node-id"() { - setup: - def callTargetsHolder = new CallTargetsHolder() - def configurer = new ConfiguredUpstreams( - Stub(FileResolver), - new UpstreamsConfig(), - Stub(CompressionConfig), - callTargetsHolder, - Mock(ApplicationEventPublisher), - Executors.newFixedThreadPool(1), - ChainsConfig.default(), - GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), - null, - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - AuthorizationConfig.default(), - new GrpcAuthContext() - ) - expect: - configurer.getHash(node, src) == expected - - where: - node | src | expected - 1 | "" | 1 - 9 | "hohoho" | 9 - null | "hohoho" | 120 - } - - def "Calculate node-id conflicting results"() { - setup: - def callTargetsHolder = new CallTargetsHolder() - def configurer = new ConfiguredUpstreams( - Stub(FileResolver), - new UpstreamsConfig(), - Stub(CompressionConfig), - callTargetsHolder, - Mock(ApplicationEventPublisher), - Executors.newFixedThreadPool(1), - ChainsConfig.default(), - GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), - null, - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - AuthorizationConfig.default(), - new GrpcAuthContext() - ) - when: - def h1 = configurer.getHash(null, "hohoho") - def h2 = configurer.getHash(null, "hohoho") - def h3 = configurer.getHash(null, "hohoho") - def h4 = configurer.getHash(null, "hohoho") - def h5 = configurer.getHash(null, "hohoho") - - then: - h1 == (byte)120 - h2 == (byte)-120 - h3 == (byte)-9 - h4 == (byte)8 - h5 == (byte)-128 - } - - def "Supporting method groups"() { - setup: - def callTargetsHolder = new CallTargetsHolder() - def configurer = new ConfiguredUpstreams( - Stub(FileResolver), - new UpstreamsConfig(), - Stub(CompressionConfig), - callTargetsHolder, - Mock(ApplicationEventPublisher), - Executors.newFixedThreadPool(1), - ChainsConfig.default(), - GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), - null, - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - Schedulers.boundedElastic(), - AuthorizationConfig.default(), - new GrpcAuthContext() - ) - def methodsGroup = new UpstreamsConfig.MethodGroups( - ["filter"] as Set, - [] as Set - ) - def upstream = new UpstreamsConfig.Upstream() - upstream.methodGroups = methodsGroup - when: - def act = configurer.buildMethods(upstream, Chain.ETHEREUM__MAINNET) - then: - act instanceof ManagedCallMethods - act.supportedMethods.findAll {it.containsIgnoreCase("filter")}.size() == 6 - } -} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy index abf68804..02d66072 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy @@ -72,7 +72,6 @@ class GenericUpstreamMock extends GenericUpstream { new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)), ChainConfig.default(), new ConnectorFactoryMock(api, new EthereumHeadMock()), - null, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&validator, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&labelDetector, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&subscriptionTopics, diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 430aeaf1..5cf1e836 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -47,12 +47,13 @@ class MultistreamHolderMock implements MultistreamHolder { upstreams[chain] = up } else if (up instanceof GenericUpstream) { upstreams[chain] = new GenericMultistream( - chain, [up as GenericUpstream], Caches.default(), + chain, Schedulers.immediate(), null, new ArrayList(), Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()) ) + upstreams[chain].addUpstream(up) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } @@ -100,7 +101,7 @@ class MultistreamHolderMock implements MultistreamHolder { Head customHead = null EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, Schedulers.boundedElastic(), + super(chain, Schedulers.immediate(), null, upstreams, caches, Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(new BraveTracer(null, null, null)), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 80c73e39..3afa1302 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -25,15 +25,17 @@ 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.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific +import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.generic.GenericMultistream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific import io.emeraldpay.etherjar.domain.BlockHash -import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import io.micrometer.core.instrument.MeterRegistry @@ -91,13 +93,15 @@ class TestingCommons { } static Multistream multistream(GenericUpstreamMock up) { - return new GenericMultistream(Chain.ETHEREUM__MAINNET, [up], Caches.default(), + return new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList(), Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()), ).tap { - start() + it.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED) + ) } } @@ -117,14 +121,14 @@ class TestingCommons { } static Multistream multistreamWithoutUpstreams(Chain chain) { - return new GenericMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), + return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { - return new GenericMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), + return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy index ebc03b62..4be9e0ee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy @@ -28,7 +28,7 @@ class CurrentMultistreamHolderSpec extends Specification { def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams()) def up = new GenericUpstreamMock("test", Chain.ETHEREUM__MAINNET, TestingCommons.api()) when: - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED)) then: current.getAvailable() == [Chain.ETHEREUM__MAINNET] current.getUpstream(Chain.ETHEREUM__MAINNET).getAll()[0] == up @@ -41,10 +41,10 @@ class CurrentMultistreamHolderSpec extends Specification { def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM_CLASSIC__MAINNET, TestingCommons.api()) def up3 = new GenericUpstreamMock("test3", Chain.ETHEREUM__MAINNET, TestingCommons.api()) when: - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)) - current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)) - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)) - current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)) then: current.getAvailable().toSet() == [Chain.ETHEREUM__MAINNET, Chain.ETHEREUM_CLASSIC__MAINNET].toSet() current.getUpstream(Chain.ETHEREUM__MAINNET).getAll().toSet() == [up1, up3].toSet() @@ -59,10 +59,10 @@ class CurrentMultistreamHolderSpec extends Specification { def up3 = new GenericUpstreamMock("test3", Chain.ETHEREUM__MAINNET, TestingCommons.api()) def up1_del = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api()) when: - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)) - current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)) - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)) - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1_del, UpstreamChangeEvent.ChangeType.REMOVED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1_del, UpstreamChangeEvent.ChangeType.REMOVED)) then: current.getAvailable().toSet() == [Chain.ETHEREUM__MAINNET, Chain.ETHEREUM_CLASSIC__MAINNET].toSet() current.getUpstream(Chain.ETHEREUM__MAINNET).getAll().toSet() == [up3].toSet() @@ -80,7 +80,7 @@ class CurrentMultistreamHolderSpec extends Specification { !act when: - current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)) + current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)) act = current.isAvailable(Chain.ETHEREUM__MAINNET) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index a7420f87..0828cbe7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -77,7 +77,6 @@ class FilteredApisSpec extends Specification { new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), ChainsConfig.ChainConfig.default(), connectorFactory, - null, cs.&validator, cs.&labelDetector, cs.&subscriptionTopics, diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index d3410349..c57129eb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -55,7 +55,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new GenericMultistream(Chain.ETHEREUM__MAINNET, [up1, up2], Caches.default(), + def aggr = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, [up1, up2], Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, @@ -190,7 +190,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 GenericMultistream(Chain.ETHEREUM__MAINNET, [up1, up2, up3], Caches.default(), + def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, [up1, up2, up3], Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, @@ -263,16 +263,16 @@ class MultistreamSpec extends Specification { setup: def up1 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"])) def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) - def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, new ArrayList(), Caches.default(), + def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList(), Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) when: - ms.onUpstreamChange( + ms.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) ) - ms.onUpstreamChange( + ms.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED) ) up1.onStatus(status(Common.AvailabilityEnum.AVAIL_UNAVAILABLE)) @@ -294,7 +294,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"])) def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) - def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, new ArrayList(), Caches.default(), + def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList(), Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, @@ -303,10 +303,10 @@ class MultistreamSpec extends Specification { def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b") def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b") when: - ms.onUpstreamChange( + ms.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) ) - ms.onUpstreamChange( + ms.processUpstreamsEvents( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED) ) def head = ms.getHead() @@ -329,25 +329,32 @@ 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 GenericMultistream(Chain.ETHEREUM__MAINNET, [up1, up2, up3], Caches.default(), + def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList(), Caches.default(), Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())) - def observer = multistream.lagObserver - multistream.onUpstreamsUpdated() + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) + ) + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED) + ) + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED) + ) expect: multistream.getAll().size() == 3 - observer.isRunning() + multistream.lagObserver.isRunning() - multistream.getAll().with { - remove(0) - remove(1) - } + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED) + ) + multistream.processUpstreamsEvents( + new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.REMOVED) + ) multistream.getAll().size() == 1 - multistream.onUpstreamsUpdated() - !observer.isRunning() multistream.lagObserver == null } @@ -360,7 +367,7 @@ class MultistreamSpec extends Specification { class TestEthereumPosMultistream extends GenericMultistream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, + super(chain, Schedulers.immediate(), null, upstreams, caches, Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()), EthereumChainSpecific.INSTANCE.&localReaderBuilder, @@ -381,11 +388,6 @@ class MultistreamSpec extends Specification { public T cast(Class selfType) { return this } - - @Override - void init() { - - } } BlockContainer createBlock(long number, String hash) { diff --git a/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt index 458fa265..dcc97552 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt @@ -16,15 +16,12 @@ import io.emeraldpay.dshackle.upstream.Upstream import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test -import org.mockito.ArgumentCaptor import org.mockito.kotlin.any import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock import org.mockito.kotlin.never -import org.mockito.kotlin.times import org.mockito.kotlin.verify import org.mockito.kotlin.whenever -import org.springframework.context.ApplicationEventPublisher import org.springframework.util.ResourceUtils import sun.misc.Signal import java.io.File @@ -38,7 +35,6 @@ class ReloadConfigTest { private val config = mock() private val reloadConfigService = ReloadConfigService(config, fileResolver, mainConfig) - private val applicationEventPublisher = mock() private val configuredUpstreams = mock() @BeforeEach @@ -66,7 +62,6 @@ class ReloadConfigTest { on { getUpstream(POLYGON__MAINNET) } doReturn msPoly } val reloadConfigUpstreamService = ReloadConfigUpstreamService( - applicationEventPublisher, currentMultistreamHolder, configuredUpstreams, ) @@ -79,8 +74,8 @@ class ReloadConfigTest { reloadConfig.handle(Signal("HUP")) - val captor = ArgumentCaptor.forClass(UpstreamChangeEvent::class.java) - verify(applicationEventPublisher, times(2)).publishEvent(captor.capture()) + verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED)) + verify(msPoly).processUpstreamsEvents(UpstreamChangeEvent(POLYGON__MAINNET, up3, UpstreamChangeEvent.ChangeType.REMOVED)) verify(configuredUpstreams).processUpstreams( UpstreamsConfig( newConfig.defaultOptions, @@ -90,14 +85,6 @@ class ReloadConfigTest { assertEquals(3, mainConfig.upstreams!!.upstreams.size) assertEquals(newConfig, mainConfig.upstreams) - assertEquals( - UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED), - captor.allValues[0], - ) - assertEquals( - UpstreamChangeEvent(POLYGON__MAINNET, up3, UpstreamChangeEvent.ChangeType.REMOVED), - captor.allValues[1], - ) } @Test @@ -123,7 +110,6 @@ class ReloadConfigTest { on { getUpstream(POLYGON__MAINNET) } doReturn msPoly } val reloadConfigUpstreamService = ReloadConfigUpstreamService( - applicationEventPublisher, currentMultistreamHolder, configuredUpstreams, ) @@ -135,8 +121,8 @@ class ReloadConfigTest { reloadConfig.handle(Signal("HUP")) - val captor = ArgumentCaptor.forClass(UpstreamChangeEvent::class.java) - verify(applicationEventPublisher, times(2)).publishEvent(captor.capture()) + verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED)) + verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.REMOVED)) verify(configuredUpstreams).processUpstreams( UpstreamsConfig( newConfig.defaultOptions, @@ -146,14 +132,6 @@ class ReloadConfigTest { verify(msEth).stop() assertEquals(1, mainConfig.upstreams!!.upstreams.size) assertEquals(newConfig, mainConfig.upstreams) - assertEquals( - UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED), - captor.allValues[0], - ) - assertEquals( - UpstreamChangeEvent(ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.REMOVED), - captor.allValues[1], - ) } @Test diff --git a/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt new file mode 100644 index 00000000..eecccb8c --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt @@ -0,0 +1,62 @@ +package io.emeraldpay.dshackle.startup.configure + +import io.emeraldpay.dshackle.BlockchainType +import io.emeraldpay.dshackle.config.UpstreamsConfig +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.Arguments +import org.junit.jupiter.params.provider.MethodSource +import org.mockito.Mockito.reset +import org.mockito.kotlin.any +import org.mockito.kotlin.mock +import org.mockito.kotlin.verify + +class UpstreamFactoryTest { + + @BeforeEach + fun beforeEach() { + reset(genericUpstreamCreator, ethereumUpstreamCreator, bitcoinUpstreamCreator) + } + + @ParameterizedTest + @MethodSource("data") + fun `create upstream based on chain`( + blockchain: BlockchainType, + verify: Runnable, + ) { + upstreamFactory.createUpstream(blockchain, UpstreamsConfig.Upstream(), emptyMap()) + + verify.run() + } + + companion object { + private val genericUpstreamCreator = mock() + private val ethereumUpstreamCreator = mock() + private val bitcoinUpstreamCreator = mock() + private val upstreamFactory = UpstreamFactory(genericUpstreamCreator, ethereumUpstreamCreator, bitcoinUpstreamCreator) + + @JvmStatic + fun data() = listOf( + Arguments.of( + BlockchainType.ETHEREUM, + Runnable { verify(ethereumUpstreamCreator).createUpstream(any(), any()) }, + ), + Arguments.of( + BlockchainType.BITCOIN, + Runnable { verify(bitcoinUpstreamCreator).createUpstream(any(), any()) }, + ), + Arguments.of( + BlockchainType.POLKADOT, + Runnable { verify(genericUpstreamCreator).createUpstream(any(), any()) }, + ), + Arguments.of( + BlockchainType.STARKNET, + Runnable { verify(genericUpstreamCreator).createUpstream(any(), any()) }, + ), + Arguments.of( + BlockchainType.UNKNOWN, + Runnable { verify(genericUpstreamCreator).createUpstream(any(), any()) }, + ), + ) + } +}