extract ethereum connectors from ethereum upstreams and add ethereum pos config

2022-07-15 18:31:19 +04:00
parent 3d44b99346
commit 92218623f2
22 changed files with 688 additions and 464 deletions
--- 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
--- 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<MappingNode>(input, "upstreams")?.value?.forEachIndexed { _, upNode ->
            val connNode = getMapping(upNode, "connection")
            if (hasAny(connNode, "ethereum")) {
-                val connConfigNode = getMapping(connNode, "ethereum")!!
-                val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>()
-                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<UpstreamsConfig.BitcoinConnection>()
-                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<String, Int>? = 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<String, Int>? = 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 <T : UpstreamsConfig.UpstreamConnection>readUpstream(config: UpstreamsConfig, upNode: MappingNode, connFactory: () -> T) {
+        val upstream = UpstreamsConfig.Upstream<T>()
+        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,
--- 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<UpstreamsConfig.EthereumPosConnection>,
+//        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<UpstreamsConfig.BitcoinConnection>,
        chain: Chain,
        options: UpstreamsConfig.Options
-    ) {
-
+    ) : Upstream? {
        val conn = config.connection!!
-        val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = 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<JsonRpcRequest, JsonRpcResponse> = 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<UpstreamsConfig.EthereumConnection>,
        chain: Chain,
        options: UpstreamsConfig.Options
-    ) {
+    ) : EthereumUpstream? {
        val conn = config.connection!!

        val urls = ArrayList<URI>()
        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<JsonRpcRequest, JsonRpcResponse>? = 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<out UpstreamsConfig.RpcConnection>): JsonRpcHttpClient? {
-        val conn = config.connection!!
-        val urls = ArrayList<URI>()
+    private fun buildHttpFactory(conn: UpstreamsConfig.RpcConnection, urls: ArrayList<URI>? = 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<URI>? = 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<URI>): 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
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt
+++ 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<JsonRpcRequest, JsonRpcResponse>
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt
+++ 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<JsonRpcRequest, JsonRpcResponse> {
+        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
+        )
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
@@ -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<JsonRpcRequest, JsonRpcResponse>,
-    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<JsonRpcRequest, JsonRpcResponse>) :
-        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<JsonRpcRequest, JsonRpcResponse> {
-        return directReader
-    }
-
-    override fun isGrpc(): Boolean {
-        return false
-    }
-
-    @Suppress("UNCHECKED_CAST")
-    override fun <T : Upstream> cast(selfType: Class<T>): T {
-        if (!selfType.isAssignableFrom(this.javaClass)) {
-            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
-        }
-        return this as T
-    }
-}
--- 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<JsonRpcRequest, JsonRpcResponse> {
+        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<UpstreamsConfig.Labels> {
        return node?.let { listOf(it.labels) } ?: emptyList()
    }
+
+    @Suppress("UNCHECKED_CAST")
+    override fun <T : Upstream> cast(selfType: Class<T>): T {
+        if (!selfType.isAssignableFrom(this.javaClass)) {
+            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
+        }
+        return this as T
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt
@@ -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<JsonRpcRequest, JsonRpcResponse> {
-        return api
-    }
-
-    override fun isGrpc(): Boolean {
-        return false
-    }
-
-    @Suppress("UNCHECKED_CAST")
-    override fun <T : Upstream> cast(selfType: Class<T>): 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
-    }
-}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt
+++ 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
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt
+++ 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<JsonRpcRequest, JsonRpcResponse>
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
+++ 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())
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
+++ 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<JsonRpcRequest, JsonRpcResponse>,
+    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<JsonRpcRequest, JsonRpcResponse> {
+        return directReader
+    }
+
+    override fun getHead(): Head {
+        return head
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
+++ 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<JsonRpcRequest, JsonRpcResponse>
+    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<JsonRpcRequest, JsonRpcResponse> {
+        return api
+    }
+
+    override fun getHead(): Head {
+        return head
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
+++ 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<Capability> {
+        return ethereumUpstream.getCapabilities()
+    }
+
+    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
+        return ethereumUpstream.getLabels()
+    }
+
+    override fun getHead() : Head {
+        return ethereumUpstream.getHead()
+    }
+
+    override fun isGrpc(): Boolean {
+        return false
+    }
+
+    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+        return ethereumUpstream.getApi()
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    override fun <T : Upstream> cast(selfType: Class<T>): T {
+        if (!selfType.isAssignableFrom(this.javaClass)) {
+            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
+        }
+        return this as T
+    }
+}
--- 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,
--- 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")
--- 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])
--- a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy
+++ 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<JsonRpcRequest, JsonRpcResponse> api
+    Head head
+
+    ConnectorFactoryMock(Reader<JsonRpcRequest, JsonRpcResponse> 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)
+    }
+}
--- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
+++ 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<JsonRpcRequest, JsonRpcResponse> api
+    Head head
+    EthereumConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
+        this.api = api
+        this.head = head
+    }
+
+    @Override
+    Reader<JsonRpcRequest, JsonRpcResponse> getApi() {
+        return this.api
+    }
+
+    @Override
+    Head getHead() {
+        return this.head
+    }
+
+    @Override
+    void start() {}
+
+    @Override
+    void stop() {}
+
+    @Override
+    boolean isRunning() {
+        return true
+    }
+}
--- 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<JsonRpcRequest, JsonRpcResponse> 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<BlockContainer> blocks) {
-        ethereumHeadMock.predefined = blocks
-    }
-
-    @Override
-    Head createHead() {
-        return ethereumHeadMock
-    }
-
-    @Override
-    Head getHead() {
-        return ethereumHeadMock
+        this.ethereumHeadMock.predefined = blocks
    }

    @Override
--- 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"])
--- a/src/test/resources/upstreams-ethereum-pos.yaml
+++ 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