solution: WS-only Ethereum upstream

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -106,6 +106,7 @@ open class UpstreamsConfig {
 
     class EthereumConnection : RpcConnection() {
         var ws: WsEndpoint? = null
+        var preferHttp: Boolean = false
     }
 
     class BitcoinConnection : RpcConnection() {

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -28,6 +28,7 @@ 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.EthereumWsFactory
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsUpstream
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -184,11 +185,6 @@ open class ConfiguredUpstreams(
                                       chain: Chain,
                                       options: UpstreamsConfig.Options) {
         val conn = config.connection!!
-        val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = buildHttpClient(config)
-        if (directApi == null) {
-            log.warn("Upstream doesn't have API configuration")
-            return
-        }
 
         val urls = ArrayList<URI>()
         val methods = buildMethods(config, chain)
@@ -209,13 +205,29 @@ open class ConfiguredUpstreams(
         }
 
         log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-        val ethereumUpstream = EthereumRpcUpstream(
+        val ethereumUpstream = if (wsFactoryApi != null && !conn.preferHttp) {
-                config.id!!,
+            EthereumWsUpstream(
-                chain, directApi, wsFactoryApi,
+                    config.id!!,
-                options, config.role,
+                    chain, wsFactoryApi,
-                QuorumForLabels.QuorumItem(1, config.labels),
+                    options, config.role,
-                methods
+                    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))
     }

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

@@ -38,11 +38,6 @@ open class EthereumRpcUpstream(
 
     private val head: Head = this.createHead()
     private var validatorSubscription: Disposable? = null
-    private val capabilities = if (options.providesBalance != false) {
-        setOf(Capability.RPC, Capability.BALANCE)
-    } else {
-        setOf(Capability.RPC)
-    }
 
     override fun setCaches(caches: Caches) {
         if (head is CachesEnabled) {
@@ -79,7 +74,7 @@ open class EthereumRpcUpstream(
 
     open fun createHead(): Head {
         return if (ethereumWsFactory != null) {
-            val ws = ethereumWsFactory.create().apply {
+            val ws = ethereumWsFactory.create(null).apply {
                 connect()
             }
             val wsHead = EthereumWsHead(ws).apply {
@@ -108,14 +103,6 @@ open class EthereumRpcUpstream(
         return directReader
     }
 
-    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
-        return listOf(node.labels)
-    }
-
-    override fun getCapabilities(): Set<Capability> {
-        return capabilities
-    }
-
     override fun isGrpc(): Boolean {
         return false
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt

@@ -26,5 +26,20 @@ abstract class EthereumUpstream(
         options: UpstreamsConfig.Options,
         role: UpstreamsConfig.UpstreamRole,
         targets: CallMethods?,
-        node: QuorumForLabels.QuorumItem?
+        private val node: QuorumForLabels.QuorumItem?
-) : DefaultUpstream(id, options, role, targets, node)
+) : DefaultUpstream(id, options, role, targets, node) {
+
+    private val capabilities = if (options.providesBalance != false) {
+        setOf(Capability.RPC, Capability.BALANCE)
+    } else {
+        setOf(Capability.RPC)
+    }
+
+    override fun getCapabilities(): Set<Capability> {
+        return capabilities
+    }
+
+    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
+        return node?.let { listOf(it.labels) } ?: emptyList()
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
+import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import io.netty.buffer.ByteBuf
@@ -58,14 +59,15 @@ class EthereumWsFactory(
 
     var basicAuth: AuthConfig.ClientBasicAuth? = null
 
-    fun create(): EthereumWs {
+    fun create(rpcMetrics: RpcMetrics?): EthereumWs {
-        return EthereumWs(uri, origin, basicAuth)
+        return EthereumWs(uri, origin, basicAuth, rpcMetrics)
     }
 
     class EthereumWs(
             private val uri: URI,
             private val origin: URI,
-            private val basicAuth: AuthConfig.ClientBasicAuth?
+            private val basicAuth: AuthConfig.ClientBasicAuth?,
+            private val rpcMetrics: RpcMetrics?
     ) : AutoCloseable {
 
         companion object {
@@ -255,12 +257,13 @@ class EthereumWsFactory(
 
         fun call(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse> {
             return Mono.fromCallable {
+                val startTime = System.nanoTime()
                 // use an internal id sequence, to avoid id conflicts with user calls
                 val internalId = sendIdSeq.getAndIncrement()
                 val originalId = originalRequest.id
-                Tuples.of(originalRequest.copy(id = internalId), originalId)
+                Tuples.of(originalRequest.copy(id = internalId), originalId, startTime)
             }.flatMap { request ->
-                waitForResponse(request.t1, request.t2)
+                waitForResponse(request.t1, request.t2, request.t3)
             }
         }
 
@@ -274,7 +277,7 @@ class EthereumWsFactory(
             }
         }
 
-        fun waitForResponse(request: JsonRpcRequest, originalId: Int): Mono<JsonRpcResponse> {
+        fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
             val expectedId = request.id.toLong()
             return Mono.just(request)
                     .flatMap {
@@ -283,6 +286,12 @@ class EthereumWsFactory(
                                 .filter { resp -> resp.id.asNumber() == expectedId }
                                 .take(1)
                                 .singleOrEmpty()
+                                .doOnNext {
+                                    rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS)
+                                }
+                                .doOnError {
+                                    rpcMetrics?.errors?.increment()
+                                }
                                 .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) }
                     }
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt

@@ -0,0 +1,120 @@
+/**
+ * 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.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: EthereumWsFactory.EthereumWs
+    private val api: JsonRpcWsClient
+
+    private var validatorSubscription: Disposable? = null
+
+    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.err")
+                        .description("Errors received on request through WebSocket JSON RPC connection")
+                        .tags(metricsTags)
+                        .register(Metrics.globalRegistry)
+        )
+
+        connection = ethereumWsFactory.create(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()
+
+        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
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -74,6 +74,32 @@ class UpstreamsConfigReaderSpec extends Specification {
         }
     }
 
+    def "Parse websocket-only config"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("upstreams-ws-only.yaml")
+        when:
+        def act = reader.read(config)
+        then:
+        act != null
+        act.upstreams.size() == 1
+        with(act.upstreams.get(0)) {
+            id == "local"
+            chain == "ethereum"
+            connection instanceof UpstreamsConfig.EthereumConnection
+            with((UpstreamsConfig.EthereumConnection) connection) {
+                rpc == null
+                ws != null
+                ws.url == new URI("ws://localhost:8546")
+                ws.basicAuth != null
+                with(ws.basicAuth) {
+                    username == "9c199ad8f281f20154fc258fe41a6814"
+                    password == "258fe4149c199ad8f2811a68f20154fc"
+                }
+            }
+        }
+    }
+
+
     def "Parse bitcoin upstreams"() {
         setup:
         def config = this.class.getClassLoader().getResourceAsStream("upstreams-bitcoin.yaml")

src/test/resources/upstreams-ws-only.yaml

@@ -0,0 +1,13 @@
+version: v1
+
+upstreams:
+  - id: local
+    chain: ethereum
+    connection:
+      ethereum:
+        ws:
+          url: "ws://localhost:8546"
+          origin: "http://localhost"
+          basic-auth:
+            username: 9c199ad8f281f20154fc258fe41a6814
+            password: 258fe4149c199ad8f2811a68f20154fc