From 92218623f20180fea3807910c5afb50d0df152dc Mon Sep 17 00:00:00 2001 From: terminal Date: Fri, 15 Jul 2022 18:31:19 +0400 Subject: [PATCH] extract ethereum connectors from ethereum upstreams and add ethereum pos config --- .../dshackle/config/UpstreamsConfig.kt | 5 + .../dshackle/config/UpstreamsConfigReader.kt | 192 ++++++++++-------- .../dshackle/startup/ConfiguredUpstreams.kt | 164 +++++++-------- .../dshackle/upstream/HttpFactory.kt | 10 + .../dshackle/upstream/HttpRpcFactory.kt | 45 ++++ .../upstream/ethereum/EthereumRpcUpstream.kt | 122 ----------- .../upstream/ethereum/EthereumUpstream.kt | 78 ++++++- .../upstream/ethereum/EthereumWsUpstream.kt | 130 ------------ .../ethereum/connectors/ConnectorFactory.kt | 10 + .../ethereum/connectors/EthereumConnector.kt | 13 ++ .../connectors/EthereumConnectorFactory.kt | 34 ++++ .../connectors/EthereumRpcConnector.kt | 76 +++++++ .../connectors/EthereumWsConnector.kt | 74 +++++++ .../ethereum_pos/EthereumPosUpstream.kt | 50 +++++ .../upstream/grpc/EthereumGrpcUpstream.kt | 8 +- .../config/UpstreamsConfigReaderSpec.groovy | 20 ++ .../dshackle/rpc/StreamHeadSpec.groovy | 4 +- .../dshackle/test/ConnectorFactoryMock.groovy | 29 +++ .../test/EthereumConnectorMock.groovy | 37 ++++ .../dshackle/test/EthereumUpstreamMock.groovy | 27 +-- .../dshackle/upstream/FilteredApisSpec.groovy | 15 +- .../resources/upstreams-ethereum-pos.yaml | 9 + 22 files changed, 688 insertions(+), 464 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy create mode 100644 src/test/resources/upstreams-ethereum-pos.yaml diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index d681a5bb..662105b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -114,6 +114,11 @@ open class UpstreamsConfig { var zeroMq: BitcoinZeroMq? = null } + class EthereumPosConnection : UpstreamConnection() { + var execution : EthereumConnection? = null + var blockPriority : Int = 0 + } + data class BitcoinZeroMq( val host: String = "127.0.0.1", val port: Int diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index a118d956..f280ec64 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -86,94 +86,16 @@ class UpstreamsConfigReader( getList(input, "upstreams")?.value?.forEachIndexed { _, upNode -> val connNode = getMapping(upNode, "connection") if (hasAny(connNode, "ethereum")) { - val connConfigNode = getMapping(connNode, "ethereum")!! - val upstream = UpstreamsConfig.Upstream() - readUpstreamCommon(upNode, upstream) - readUpstreamStandard(upNode, upstream) - if (isValid(upstream)) { - config.upstreams.add(upstream) - val connection = UpstreamsConfig.EthereumConnection() - upstream.connection = connection - getMapping(connConfigNode, "rpc")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) - connection.rpc = http - http.basicAuth = authConfigReader.readClientBasicAuth(node) - http.tls = authConfigReader.readClientTls(node) - } - } - getMapping(connConfigNode, "ws")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val ws = UpstreamsConfig.WsEndpoint(URI(url)) - connection.ws = ws - getValueAsString(node, "origin")?.let { origin -> - ws.origin = URI(origin) - } - ws.basicAuth = authConfigReader.readClientBasicAuth(node) - - getValueAsBytes(node, "frameSize")?.let { - if (it < 65_535) { - throw IllegalStateException("frameSize cannot be less than 64Kb") - } - ws.frameSize = it - } - getValueAsBytes(node, "msgSize")?.let { - if (it < 65_535) { - throw IllegalStateException("msgSize cannot be less than 64Kb") - } - ws.msgSize = it - } - } - } - } else { - log.error("Upstream at #0 has invalid configuration") + readUpstream(config, upNode) { + readEthereumConnection(getMapping(connNode, "ethereum")!!) } } else if (hasAny(connNode, "bitcoin")) { - val connConfigNode = getMapping(connNode, "bitcoin")!! - val upstream = UpstreamsConfig.Upstream() - readUpstreamCommon(upNode, upstream) - readUpstreamStandard(upNode, upstream) - if (isValid(upstream)) { - config.upstreams.add(upstream) - val connection = UpstreamsConfig.BitcoinConnection() - upstream.connection = connection - getMapping(connConfigNode, "rpc")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) - connection.rpc = http - http.basicAuth = authConfigReader.readClientBasicAuth(node) - http.tls = authConfigReader.readClientTls(node) - } - } - getMapping(connConfigNode, "esplora")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) - http.basicAuth = authConfigReader.readClientBasicAuth(node) - http.tls = authConfigReader.readClientTls(node) - connection.esplora = http - } - } - getMapping(connConfigNode, "zeromq")?.let { node -> - getValueAsString(node, "address")?.let { address -> - val zmqConfig: Pair? = try { - if (address.contains(":")) { - address.split(":").let { - Pair(it[0], it[1].toInt()) - } - } else { - Pair("127.0.0.1", address.toInt()) - } - } catch (t: Throwable) { - log.warn("Invalid config for ZeroMQ: $address. Expected to be in format HOST:PORT") - null - } - zmqConfig?.let { - connection.zeroMq = UpstreamsConfig.BitcoinZeroMq(it.first, it.second) - } - } - } - } else { - log.error("Upstream at #0 has invalid configuration") + readUpstream(config, upNode) { + readBitcoinConnection(getMapping(connNode, "bitcoin")!!) + } + } else if (hasAny(connNode, "ethereum-pos")) { + readUpstream(config, upNode) { + readEthereumPosConnection(getMapping(connNode, "ethereum-pos")!!) } } else if (hasAny(connNode, "grpc")) { val connConfigNode = getMapping(connNode, "grpc")!! @@ -200,6 +122,104 @@ class UpstreamsConfigReader( return config } + private fun readBitcoinConnection(connConfigNode: MappingNode): UpstreamsConfig.BitcoinConnection { + val connection = UpstreamsConfig.BitcoinConnection() + getMapping(connConfigNode, "rpc")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val http = UpstreamsConfig.HttpEndpoint(URI(url)) + connection.rpc = http + http.basicAuth = authConfigReader.readClientBasicAuth(node) + http.tls = authConfigReader.readClientTls(node) + } + } + getMapping(connConfigNode, "esplora")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val http = UpstreamsConfig.HttpEndpoint(URI(url)) + http.basicAuth = authConfigReader.readClientBasicAuth(node) + http.tls = authConfigReader.readClientTls(node) + connection.esplora = http + } + } + getMapping(connConfigNode, "zeromq")?.let { node -> + getValueAsString(node, "address")?.let { address -> + val zmqConfig: Pair? = try { + if (address.contains(":")) { + address.split(":").let { + Pair(it[0], it[1].toInt()) + } + } else { + Pair("127.0.0.1", address.toInt()) + } + } catch (t: Throwable) { + log.warn("Invalid config for ZeroMQ: $address. Expected to be in format HOST:PORT") + null + } + zmqConfig?.let { + connection.zeroMq = UpstreamsConfig.BitcoinZeroMq(it.first, it.second) + } + } + } + return connection + } + + private fun readEthereumPosConnection(connConfigNode: MappingNode) : UpstreamsConfig.EthereumPosConnection { + val connection = UpstreamsConfig.EthereumPosConnection() + getMapping(connConfigNode, "execution")?.let { + connection.execution = readEthereumConnection(it) + } + getValueAsInt(connConfigNode, "block-priority")?.let { + connection.blockPriority = it + } + return connection + } + private fun readEthereumConnection(connConfigNode : MappingNode) : UpstreamsConfig.EthereumConnection { + val connection = UpstreamsConfig.EthereumConnection() + getMapping(connConfigNode, "rpc")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val http = UpstreamsConfig.HttpEndpoint(URI(url)) + connection.rpc = http + http.basicAuth = authConfigReader.readClientBasicAuth(node) + http.tls = authConfigReader.readClientTls(node) + } + } + getMapping(connConfigNode, "ws")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val ws = UpstreamsConfig.WsEndpoint(URI(url)) + connection.ws = ws + getValueAsString(node, "origin")?.let { origin -> + ws.origin = URI(origin) + } + ws.basicAuth = authConfigReader.readClientBasicAuth(node) + + getValueAsBytes(node, "frameSize")?.let { + if (it < 65_535) { + throw IllegalStateException("frameSize cannot be less than 64Kb") + } + ws.frameSize = it + } + getValueAsBytes(node, "msgSize")?.let { + if (it < 65_535) { + throw IllegalStateException("msgSize cannot be less than 64Kb") + } + ws.msgSize = it + } + } + } + return connection + } + + private fun readUpstream(config: UpstreamsConfig, upNode: MappingNode, connFactory: () -> T) { + val upstream = UpstreamsConfig.Upstream() + readUpstreamCommon(upNode, upstream) + readUpstreamStandard(upNode, upstream) + if (isValid(upstream)) { + config.upstreams.add(upstream) + upstream.connection = connFactory() + } else { + log.error("Upstream at #0 has invalid configuration") + } + } + fun isValid(upstream: UpstreamsConfig.Upstream<*>): Boolean { val id = upstream.id // In general, we just check that id is suitable for urls and references, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 48e4f189..bcf9e216 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -20,9 +20,7 @@ import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead @@ -31,20 +29,14 @@ 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.EthereumRpcUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory -import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsUpstream +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.Chain -import io.micrometer.core.instrument.Counter -import io.micrometer.core.instrument.Metrics -import io.micrometer.core.instrument.Tag -import io.micrometer.core.instrument.Timer import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Repository @@ -83,18 +75,24 @@ open class ConfiguredUpstreams( } val options = (up.options ?: UpstreamsConfig.Options()) .merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults()) - when (BlockchainType.from(chain)) { + val upstream = when (BlockchainType.from(chain)) { BlockchainType.ETHEREUM -> { buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options) } BlockchainType.BITCOIN -> { buildBitcoinUpstream(up.cast(UpstreamsConfig.BitcoinConnection::class.java), chain, options) } +// BlockchainType.ETHEREUM_POS -> { +// buildEthereumPosUpstream(up.cast(UpstreamsConfig.EthereumPosConnection::class.java), chain, options) +// } else -> { log.error("Chain is unsupported: ${up.chain}") return@forEach } } + upstream?.let { + currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED)) + } } } } @@ -141,19 +139,33 @@ open class ConfiguredUpstreams( } } +// private fun buildEthereumPosUpstream( +// config: UpstreamsConfig.Upstream, +// chain: Chain, +// options: UpstreamsConfig.Options +// ) : Upstream? { +// val conn = config.connection!! +// val execution = conn.execution +// if (execution == null) { +// log.warn("Upstream doesn't have execution layer configuration") +// return null +// } +// +// val connectorFactory = buildEthereumConnectorFactory(execution, chain) +// } + private fun buildBitcoinUpstream( config: UpstreamsConfig.Upstream, chain: Chain, options: UpstreamsConfig.Options - ) { - + ) : Upstream? { val conn = config.connection!! - val directApi: Reader? = buildHttpClient(config) - if (directApi == null) { + val httpFactory = buildHttpFactory(conn) + if (httpFactory == null) { log.warn("Upstream doesn't have API configuration") - return + return null } - + val directApi: Reader = httpFactory.create(config.id, chain) val esplora = conn.esplora?.let { endpoint -> val tls = endpoint.tls?.let { tls -> tls.ca?.let { ca -> @@ -180,63 +192,34 @@ open class ConfiguredUpstreams( QuorumForLabels.QuorumItem(1, config.labels), methods, esplora ) - upstream.start() - currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED)) + return upstream } private fun buildEthereumUpstream( config: UpstreamsConfig.Upstream, chain: Chain, options: UpstreamsConfig.Options - ) { + ) : EthereumUpstream? { val conn = config.connection!! val urls = ArrayList() val methods = buildMethods(config, chain) - conn.rpc?.let { endpoint -> - urls.add(endpoint.url) - } - val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint -> - val wsApi = EthereumWsFactory( - endpoint.url, - endpoint.origin ?: URI("http://localhost"), - ) - wsApi.config = endpoint - endpoint.basicAuth?.let { auth -> - wsApi.basicAuth = auth - } - urls.add(endpoint.url) - wsApi + val connectorFactory = buildEthereumConnectorFactory(conn, chain, urls) + if (connectorFactory == null) { + return null } - - log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val ethereumUpstream = if (wsFactoryApi != null && !conn.preferHttp) { - EthereumWsUpstream( - config.id!!, - chain, wsFactoryApi, - options, config.role, - QuorumForLabels.QuorumItem(1, config.labels), - methods - ) - } else { - val directApi: Reader? = buildHttpClient(config) - if (directApi == null) { - log.warn("Upstream doesn't have API configuration") - return - } - EthereumRpcUpstream( - config.id!!, - chain, directApi, wsFactoryApi, - options, config.role, - QuorumForLabels.QuorumItem(1, config.labels), - methods - ) - } - - ethereumUpstream.start() - currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED)) + val upstream = EthereumUpstream( + config.id!!, + chain, + options, config.role, + methods, + QuorumForLabels.QuorumItem(1, config.labels), + connectorFactory + ) + upstream.start() + return upstream } private fun buildGrpcUpstream( @@ -262,39 +245,42 @@ open class ConfiguredUpstreams( .subscribe(currentUpstreams::update) } - private fun buildHttpClient(config: UpstreamsConfig.Upstream): JsonRpcHttpClient? { - val conn = config.connection!! - val urls = ArrayList() + private fun buildHttpFactory(conn: UpstreamsConfig.RpcConnection, urls: ArrayList? = null): HttpRpcFactory? { return conn.rpc?.let { endpoint -> val tls = conn.rpc?.tls?.let { tls -> tls.ca?.let { ca -> fileResolver.resolve(ca).readBytes() } } - val metricsTags = listOf( - // "unknown" is not supposed to happen - Tag.of("upstream", config.id ?: "unknown"), - // UNSPECIFIED shouldn't happen too - Tag.of("chain", (Global.chainById(config.chain).chainCode)) - ) - val metrics = RpcMetrics( - Timer.builder("upstream.rpc.conn") - .description("Request time through a HTTP JSON RPC connection") - .tags(metricsTags) - .publishPercentileHistogram() - .register(Metrics.globalRegistry), - Counter.builder("upstream.rpc.fail") - .description("Number of failures of HTTP JSON RPC requests") - .tags(metricsTags) - .register(Metrics.globalRegistry) - ) - urls.add(endpoint.url) - JsonRpcHttpClient( - endpoint.url.toString(), - metrics, - conn.rpc?.basicAuth, - tls - ) + urls?.add(endpoint.url) + HttpRpcFactory(endpoint.url.toString(), conn.rpc?.basicAuth, tls) } } + + private fun buildWsFactory(conn: UpstreamsConfig.EthereumConnection, urls: ArrayList? = null): EthereumWsFactory? { + return conn.ws?.let { endpoint -> + val wsApi = EthereumWsFactory( + endpoint.url, + endpoint.origin ?: URI("http://localhost"), + ) + wsApi.config = endpoint + endpoint.basicAuth?.let { auth -> + wsApi.basicAuth = auth + } + urls?.add(endpoint.url) + wsApi + } + } + + private fun buildEthereumConnectorFactory(conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList): EthereumConnectorFactory? { + val wsFactoryApi = buildWsFactory(conn, urls) + val httpFactory = buildHttpFactory(conn, urls) + log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") + val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory) + if (!connectorFactory.isValid()) { + log.warn("Upstream configuration is invalid (probably no http endpoint)") + return null + } + return connectorFactory + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt new file mode 100644 index 00000000..10d99560 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt @@ -0,0 +1,10 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain + +interface HttpFactory { + fun create(id: String?, chain: Chain): Reader +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt new file mode 100644 index 00000000..9261cc6c --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt @@ -0,0 +1,45 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.config.AuthConfig +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag +import io.micrometer.core.instrument.Timer + +open class HttpRpcFactory( + private val url: String, + private val basicAuth: AuthConfig.ClientBasicAuth?, + private val tls: ByteArray? +) : HttpFactory { + override fun create(id: String?, chain: Chain): Reader { + val metricsTags = listOf( + // "unknown" is not supposed to happen + Tag.of("upstream", id ?: "unknown"), + // UNSPECIFIED shouldn't happen too + Tag.of("chain", chain.chainCode) + ) + val metrics = RpcMetrics( + Timer.builder("upstream.rpc.conn") + .description("Request time through a HTTP JSON RPC connection") + .tags(metricsTags) + .publishPercentileHistogram() + .register(Metrics.globalRegistry), + Counter.builder("upstream.rpc.fail") + .description("Number of failures of HTTP JSON RPC requests") + .tags(metricsTags) + .register(Metrics.globalRegistry) + ) + return JsonRpcHttpClient( + url, + metrics, + basicAuth, + tls + ) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt deleted file mode 100644 index 51bed8d7..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ /dev/null @@ -1,122 +0,0 @@ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.cache.CachesEnabled -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.MergedHead -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.grpc.Chain -import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle -import reactor.core.Disposable -import java.time.Duration - -open class EthereumRpcUpstream( - id: String, - val chain: Chain, - private val directReader: Reader, - private val ethereumWsFactory: EthereumWsFactory? = null, - options: UpstreamsConfig.Options, - role: UpstreamsConfig.UpstreamRole, - private val node: QuorumForLabels.QuorumItem, - targets: CallMethods -) : EthereumUpstream(id, options, role, targets, node), Upstream, CachesEnabled, Lifecycle { - - constructor(id: String, chain: Chain, api: Reader) : - this( - id, chain, api, null, - UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.PRIMARY, - QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), - DirectCallMethods() - ) - - private val log = LoggerFactory.getLogger(EthereumRpcUpstream::class.java) - - private val head: Head = this.createHead() - private var validatorSubscription: Disposable? = null - - override fun setCaches(caches: Caches) { - if (head is CachesEnabled) { - head.setCaches(caches) - } - } - - override fun start() { - log.info("Configured for ${chain.chainName}") - - if (getOptions().disableValidation != null && getOptions().disableValidation!!) { - log.warn("Disable validation for upstream ${this.getId()}") - this.setLag(0) - this.setStatus(UpstreamAvailability.OK) - } else { - log.debug("Start validation for upstream ${this.getId()}") - val validator = EthereumUpstreamValidator(this, getOptions()) - validatorSubscription = validator.start() - .subscribe(this::setStatus) - } - } - - override fun isRunning(): Boolean { - return true - } - - override fun stop() { - validatorSubscription?.dispose() - validatorSubscription = null - if (head is Lifecycle) { - head.stop() - } - } - - open fun createHead(): Head { - return if (ethereumWsFactory != null) { - // do not set upstream to the WS, since it doesn't control the RPC upstream - val ws = ethereumWsFactory.create(null, null, null).apply { - connect() - } - val wsHead = EthereumWsHead(ws).apply { - start() - } - // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed - val rpcHead = EthereumRpcHead(getApi(), Duration.ofSeconds(60)).apply { - start() - } - MergedHead(listOf(rpcHead, wsHead)).apply { - start() - } - } else { - log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC") - EthereumRpcHead(getApi()).apply { - start() - } - } - } - - override fun getHead(): Head { - return head - } - - override fun getApi(): Reader { - return directReader - } - - override fun isGrpc(): Boolean { - return false - } - - @Suppress("UNCHECKED_CAST") - override fun cast(selfType: Class): T { - if (!selfType.isAssignableFrom(this.javaClass)) { - throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") - } - return this as T - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index c14bd120..4248f78f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -16,19 +16,79 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Capability -import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable -abstract class EthereumUpstream( +open class EthereumUpstream( id: String, + val chain: Chain, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, - private val node: QuorumForLabels.QuorumItem? -) : DefaultUpstream(id, options, role, targets, node) { + private val node: QuorumForLabels.QuorumItem?, + connectorFactory: ConnectorFactory +) : DefaultUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled { + private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) + private val validator : EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions()) + private val connector : EthereumConnector = connectorFactory.create(this, validator, chain) + + private var validatorSubscription: Disposable? = null + + override fun setCaches(caches: Caches) { + if (connector is CachesEnabled) { + connector.setCaches(caches) + } + } + + override fun start() { + log.info("Configured for ${chain.chainName}") + connector.start() + if (getOptions().disableValidation != null && getOptions().disableValidation!!) { + log.warn("Disable validation for upstream ${this.getId()}") + this.setLag(0) + this.setStatus(UpstreamAvailability.OK) + } else { + log.debug("Start validation for upstream ${this.getId()}") + val validator = EthereumUpstreamValidator(this, getOptions()) + validatorSubscription = validator.start() + .subscribe(this::setStatus) + } + } + override fun getHead(): Head { + return connector.getHead() + } + + override fun stop() { + validatorSubscription?.dispose() + validatorSubscription = null + connector.stop() + } + + override fun isRunning(): Boolean { + return connector.isRunning + } + + override fun getApi(): Reader { + return connector.getApi() + } + + override fun isGrpc(): Boolean { + return false + } private val capabilities = if (options.providesBalance != false) { setOf(Capability.RPC, Capability.BALANCE) @@ -43,4 +103,12 @@ abstract class EthereumUpstream( override fun getLabels(): Collection { return node?.let { listOf(it.labels) } ?: emptyList() } + + @Suppress("UNCHECKED_CAST") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt deleted file mode 100644 index c9ddb32b..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ /dev/null @@ -1,130 +0,0 @@ -/** - * Copyright (c) 2021 EmeraldPay, Inc - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient -import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics -import io.emeraldpay.grpc.Chain -import io.micrometer.core.instrument.Counter -import io.micrometer.core.instrument.Metrics -import io.micrometer.core.instrument.Tag -import io.micrometer.core.instrument.Timer -import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle -import reactor.core.Disposable - -class EthereumWsUpstream( - id: String, - val chain: Chain, - ethereumWsFactory: EthereumWsFactory, - options: UpstreamsConfig.Options, - role: UpstreamsConfig.UpstreamRole, - node: QuorumForLabels.QuorumItem, - targets: CallMethods -) : EthereumUpstream(id, options, role, targets, node), Upstream, Lifecycle { - - companion object { - private val log = LoggerFactory.getLogger(EthereumWsUpstream::class.java) - } - - private val head: EthereumWsHead - private val connection: WsConnection - private val api: JsonRpcWsClient - - private var validatorSubscription: Disposable? = null - private val validator: EthereumUpstreamValidator - - init { - val metricsTags = listOf( - Tag.of("upstream", id), - // UNSPECIFIED shouldn't happen too - Tag.of("chain", chain.chainCode) - ) - val metrics = RpcMetrics( - Timer.builder("upstream.ws.conn") - .description("Request time through a WebSocket JSON RPC connection") - .tags(metricsTags) - .publishPercentileHistogram() - .register(Metrics.globalRegistry), - Counter.builder("upstream.ws.fail") - .description("Number of failures of WebSocket JSON RPC requests") - .tags(metricsTags) - .register(Metrics.globalRegistry) - ) - - validator = EthereumUpstreamValidator(this, getOptions()) - - connection = ethereumWsFactory.create(this, validator, metrics) - head = EthereumWsHead(connection) - api = JsonRpcWsClient(connection) - } - - override fun getHead(): Head { - return head - } - - override fun getApi(): Reader { - return api - } - - override fun isGrpc(): Boolean { - return false - } - - @Suppress("UNCHECKED_CAST") - override fun cast(selfType: Class): T { - if (!selfType.isAssignableFrom(this.javaClass)) { - throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") - } - return this as T - } - - override fun start() { - connection.connect() - head.start() - - if (getOptions().disableValidation != null && getOptions().disableValidation!!) { - log.warn("Disable validation for upstream ${this.getId()}") - this.setLag(0) - this.setStatus(UpstreamAvailability.OK) - } else { - log.debug("Start validation for upstream ${this.getId()}") - val validator = EthereumUpstreamValidator(this, getOptions()) - validatorSubscription = validator.start() - .subscribe(this::setStatus) - } - } - - override fun stop() { - validatorSubscription?.dispose() - validatorSubscription = null - head.stop() - connection.close() - } - - override fun isRunning(): Boolean { - return head.isRunning - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt new file mode 100644 index 00000000..09c5f26d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt @@ -0,0 +1,10 @@ +package io.emeraldpay.dshackle.upstream.ethereum.connectors + +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator +import io.emeraldpay.grpc.Chain + +interface ConnectorFactory { + fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector + fun isValid(): Boolean +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt new file mode 100644 index 00000000..dcd8f5b0 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt @@ -0,0 +1,13 @@ +package io.emeraldpay.dshackle.upstream.ethereum.connectors + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import org.springframework.context.Lifecycle + +interface EthereumConnector : Lifecycle { + fun getHead(): Head + + fun getApi(): Reader +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt new file mode 100644 index 00000000..68c11b43 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt @@ -0,0 +1,34 @@ +package io.emeraldpay.dshackle.upstream.ethereum.connectors + +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.HttpFactory +import io.emeraldpay.dshackle.upstream.HttpRpcFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory + +open class EthereumConnectorFactory( + private val preferHttp: Boolean, + private val wsFactory: EthereumWsFactory?, + private val httpFactory: HttpFactory? +): ConnectorFactory { + private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java) + + override fun isValid(): Boolean { + if (preferHttp && httpFactory == null) { + return false; + } + return true + } + + override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector { + if (wsFactory!= null && !preferHttp) { + return EthereumWsConnector(wsFactory, upstream, validator, chain) + } + if (httpFactory == null) { + throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set") + } + return EthereumRpcConnector(httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId()) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt new file mode 100644 index 00000000..1dca02ba --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -0,0 +1,76 @@ +package io.emeraldpay.dshackle.upstream.ethereum.connectors + +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.ethereum.* +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import java.time.Duration + +class EthereumRpcConnector( + private val directReader : Reader, + wsFactory: EthereumWsFactory?, + id : String, +) : EthereumConnector, CachesEnabled { + private val conn : WsConnection? + private val head : Head + + companion object { + private val log = LoggerFactory.getLogger(EthereumRpcConnector::class.java) + } + + init { + if (wsFactory != null) { + // do not set upstream to the WS, since it doesn't control the RPC upstream + conn = wsFactory.create(null, null, null) + val wsHead = EthereumWsHead(conn) + // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed + val rpcHead = EthereumRpcHead(directReader, Duration.ofSeconds(60)) + head = MergedHead(listOf(rpcHead, wsHead)) + } else { + conn = null + log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") + head = EthereumRpcHead(directReader) + } + } + + override fun setCaches(caches: Caches) { + if (head is CachesEnabled) { + head.setCaches(caches) + } + } + + override fun start() { + if (head is Lifecycle) { + head.start() + } + conn?.connect() + } + + override fun isRunning(): Boolean { + if (head is Lifecycle) { + return head.isRunning + } + return true + } + + override fun stop() { + if (head is Lifecycle) { + head.stop() + } + conn?.close() + } + + override fun getApi(): Reader { + return directReader + } + + override fun getHead(): Head { + return head + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt new file mode 100644 index 00000000..fdb2e834 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -0,0 +1,74 @@ +package io.emeraldpay.dshackle.upstream.ethereum.connectors + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead +import io.emeraldpay.dshackle.upstream.ethereum.WsConnection +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient +import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag +import io.micrometer.core.instrument.Timer + +class EthereumWsConnector( + wsFactory: EthereumWsFactory, + upstream: DefaultUpstream, + validator: EthereumUpstreamValidator, + chain: Chain, +) : EthereumConnector { + private val conn: WsConnection + private val api: Reader + private val head: EthereumWsHead + + init { + val metricsTags = listOf( + Tag.of("upstream", upstream.getId()), + // UNSPECIFIED shouldn't happen too + Tag.of("chain", chain.chainCode) + ) + val metrics = RpcMetrics( + Timer.builder("upstream.ws.conn") + .description("Request time through a WebSocket JSON RPC connection") + .tags(metricsTags) + .publishPercentileHistogram() + .register(Metrics.globalRegistry), + Counter.builder("upstream.ws.fail") + .description("Number of failures of WebSocket JSON RPC requests") + .tags(metricsTags) + .register(Metrics.globalRegistry) + ) + + conn = wsFactory.create(upstream, validator, metrics) + head = EthereumWsHead(conn) + api = JsonRpcWsClient(conn) + } + + override fun start() { + conn.connect() + head.start() + } + + override fun isRunning(): Boolean { + return head.isRunning + } + + override fun stop() { + conn.close() + head.stop() + } + + override fun getApi(): Reader { + return api + } + + override fun getHead(): Head { + return head + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt new file mode 100644 index 00000000..298b784a --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt @@ -0,0 +1,50 @@ +package io.emeraldpay.dshackle.upstream.ethereum_pos + +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.Capability +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse + +class EthereumPosUpstream( + id: String, + options: UpstreamsConfig.Options, + role: UpstreamsConfig.UpstreamRole, + targets: CallMethods?, + node: QuorumForLabels.QuorumItem?, + private val ethereumUpstream: EthereumUpstream +) : DefaultUpstream(id, options, role, targets, node) { + override fun getCapabilities(): Set { + return ethereumUpstream.getCapabilities() + } + + override fun getLabels(): Collection { + return ethereumUpstream.getLabels() + } + + override fun getHead() : Head { + return ethereumUpstream.getHead() + } + + override fun isGrpc(): Boolean { + return false + } + + override fun getApi(): Reader { + return ethereumUpstream.getApi() + } + + @Suppress("UNCHECKED_CAST") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 3951d981..7a3c5006 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -24,11 +24,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Capability -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient @@ -53,7 +49,7 @@ open class EthereumGrpcUpstream( private val chain: Chain, private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: JsonRpcGrpcClient -) : EthereumUpstream( +) : DefaultUpstream( "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", UpstreamsConfig.Options.getDefaults(), role, diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 73f8db20..d69c22b6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -154,6 +154,26 @@ class UpstreamsConfigReaderSpec extends Specification { } } + def "Parse ethereum pos upstreams"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-ethereum-pos.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 1 + with(act.upstreams.get(0)) { + id == "eth2-1" + chain == "ropsten" + connection instanceof UpstreamsConfig.EthereumPosConnection + with((UpstreamsConfig.EthereumPosConnection) connection) { + execution.rpc != null + execution.rpc.url == new URI("http://34.106.60.110:8545") + blockPriority == 100 + } + } + } + def "Parse bitcoin upstreams with esplora"() { setup: def config = this.class.getClassLoader().getResourceAsStream("upstreams-bitcoin-esplora.yaml") diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy index ae121f21..068d4240 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy @@ -87,7 +87,9 @@ class StreamHeadSpec extends Specification { ) then: StepVerifier.create(flux.take(2)) - .then { upstream.nextBlock(BlockContainer.from(blocks[0])) } + .then { + upstream.nextBlock(BlockContainer.from(blocks[0])) + } .expectNext(heads[0]) .then { upstream.nextBlock(BlockContainer.from(blocks[1])) } .expectNext(heads[1]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy new file mode 100644 index 00000000..34d7e3d6 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy @@ -0,0 +1,29 @@ +package io.emeraldpay.dshackle.test + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator +import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain + +class ConnectorFactoryMock implements ConnectorFactory { + Reader api + Head head + + ConnectorFactoryMock(Reader api, Head head) { + this.api = api + this.head = head + } + + boolean isValid() { + return true + } + + EthereumConnector create(DefaultUpstream upstream, EthereumUpstreamValidator validator, Chain chain) { + return new EthereumConnectorMock(api, head) + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy new file mode 100644 index 00000000..5e25d636 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy @@ -0,0 +1,37 @@ +package io.emeraldpay.dshackle.test + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse + +class EthereumConnectorMock implements EthereumConnector { + Reader api + Head head + EthereumConnectorMock(Reader api, Head head) { + this.api = api + this.head = head + } + + @Override + Reader getApi() { + return this.api + } + + @Override + Head getHead() { + return this.head + } + + @Override + void start() {} + + @Override + void stop() {} + + @Override + boolean isRunning() { + return true + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 4b292dd7..43caf2ee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -19,14 +19,12 @@ package io.emeraldpay.dshackle.test import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -36,9 +34,10 @@ import io.emeraldpay.grpc.Chain import org.jetbrains.annotations.NotNull import org.reactivestreams.Publisher -class EthereumUpstreamMock extends EthereumRpcUpstream { - EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() +class EthereumUpstreamMock extends EthereumUpstream { + EthereumHeadMock ethereumHeadMock + static CallMethods allMethods() { new AggregatedCallMethods([ @@ -61,32 +60,24 @@ class EthereumUpstreamMock extends EthereumRpcUpstream { } EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods) { - super(id, chain, api, null, + super(id, chain, UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.PRIMARY, + methods, new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), - methods) + new ConnectorFactoryMock(api, new EthereumHeadMock())) + this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) setStatus(UpstreamAvailability.OK) start() } void nextBlock(BlockContainer block) { - ethereumHeadMock.nextBlock(block) + this.ethereumHeadMock.nextBlock(block) } void setBlocks(Publisher blocks) { - ethereumHeadMock.predefined = blocks - } - - @Override - Head createHead() { - return ethereumHeadMock - } - - @Override - Head getHead() { - return ethereumHeadMock + this.ethereumHeadMock.predefined = blocks } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 960e0e40..eefdf46a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -21,11 +21,9 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.test.EthereumApiStub import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.grpc.Chain -import reactor.core.publisher.Flux import reactor.test.StepVerifier import spock.lang.Retry import spock.lang.Specification @@ -46,15 +44,18 @@ class FilteredApisSpec extends Specification { [test: "foo"], [test: "baz"] ].collect { - new EthereumRpcUpstream( + def httpFactory = Mock(HttpFactory) { + create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" } + } + def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory) + new EthereumUpstream( "test", Chain.ETHEREUM, - TestingCommons.api().tap { it.id = "${i++}" }, - (EthereumWsFactory) null, new UpstreamsConfig.Options(), UpstreamsConfig.UpstreamRole.PRIMARY, + ethereumTargets, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), - ethereumTargets + connectorFactory ) } def matcher = new Selector.LabelMatcher("test", ["foo"]) diff --git a/src/test/resources/upstreams-ethereum-pos.yaml b/src/test/resources/upstreams-ethereum-pos.yaml new file mode 100644 index 00000000..b2a1a180 --- /dev/null +++ b/src/test/resources/upstreams-ethereum-pos.yaml @@ -0,0 +1,9 @@ + upstreams: + - id: eth2-1 + chain: ropsten + connection: + ethereum-pos: + execution: + rpc: + url: "http://34.106.60.110:8545" + block-priority: 100 \ No newline at end of file