Merge pull request #114 from emeraldpay/feat/calls-with-ws

fix #107

.github/workflows/test.yaml

@@ -32,6 +32,8 @@ jobs:
         uses: eskatos/gradle-command-action@v1
         with:
           arguments: check
+        env:
+          CI: true
 
       - name: Upload Coverage Report
         uses: codecov/codecov-action@v1
@@ -64,6 +66,7 @@ jobs:
         with:
           arguments: check
         env:
+          CI: true
           DSHACKLE_TEST_ENABLED: redis
           REDIS_HOST: redis
           REDIS_PORT: 6379

build.gradle

@@ -132,6 +132,7 @@ dependencies {
     testImplementation "io.projectreactor:reactor-test:$reactorVersion"
     testImplementation 'org.objenesis:objenesis:3.1'
     testImplementation 'org.mock-server:mockserver-netty:5.11.2'
+    testImplementation "org.java-websocket:Java-WebSocket:1.5.1"
     testImplementation "nl.jqno.equalsverifier:equalsverifier:3.3"
     testImplementation "org.codehaus.groovy:groovy:${groovyVersion}"
 }

docs/reference-configuration.adoc

@@ -527,6 +527,8 @@ configuration, and may be omitted for most of the situations.
         basic-auth:
           username: 9c199ad8f281f20154fc258fe41a6814
           password: 258fe4149c199ad8f2811a68f20154fc
+        frameSize: 5mb
+        msgSize: 15mb
 ----
 
 .Main Config
@@ -592,7 +594,7 @@ Example: `https://kovan.infura.io/v3/${INFURA_USER}`
 
 | `rpc.basic-auth` + `rpc.basic-auth.username`, `rpc.basic-auth.password`
 a| HTTP Basic Auth configuration, if required by the remote server. +
-   Values can also reference env variables, for example:
+Values can also reference env variables, for example:
 [source,yaml]
 ----
 rpc:
@@ -603,7 +605,8 @@ rpc:
 ----
 
 | `ws.url`
-| Websocket URL to connect to. Optional, but optimizes performance if it's available.
+| Websocket URL to connect to.
+Optional, but optimizes performance if it's available.
 
 | `ws.origin`
 | HTTP `Origin` if required by Websocket remote server.
@@ -611,6 +614,15 @@ rpc:
 | `ws.basic-auth` + ...
 | Websocket Basic Auth configuration, if required by the remote server
 
+| `ws.frameSize`
+| WebSocket frame size limit.
+Ex `1kb`, `1024` (same as `1kb), `2mb`, etc.
+Default is 5Mb
+
+| `ws.msgSize`
+| Total limit for a message size consisting from multiple frames.
+Ex `1kb`, `1024` (same as `1kb), `2mb`, etc.
+Default is 15Mb
 
 |===
 

src/main/kotlin/io/emeraldpay/dshackle/Global.kt

@@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspentDeserializer
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspentDeserializer
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import java.text.SimpleDateFormat
 import java.util.*
@@ -48,6 +49,7 @@ class Global {
 
             module.addDeserializer(EsploraUnspent::class.java, EsploraUnspentDeserializer())
             module.addDeserializer(RpcUnspent::class.java, RpcUnspentDeserializer())
+            module.addDeserializer(JsonRpcRequest::class.java, JsonRpcRequest.Deserializer())
 
             val objectMapper = ObjectMapper()
             objectMapper.registerModule(module)

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() {
@@ -120,10 +121,11 @@ open class UpstreamsConfig {
     class WsEndpoint(val url: URI) {
         var origin: URI? = null
         var basicAuth: AuthConfig.ClientBasicAuth? = null
+        var frameSize: Int? = null
+        var msgSize: Int? = null
     }
 
 
-
     //TODO make it unmodifiable after initial load
     class Labels: HashMap<String, String>() {
 

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

@@ -112,6 +112,19 @@ class UpstreamsConfigReader(
                                 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 {

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

@@ -127,6 +127,22 @@ abstract class YamlConfigReader {
         }
     }
 
+    fun getValueAsBytes(mappingNode: MappingNode?, key: String): Int? {
+        return getValueAsString(mappingNode, key)?.let(envVariables::postProcess)?.let {
+            val m = Regex("^(\\d+)(m|mb|k|kb|b)?$").find(it.lowercase().trim())
+                    ?: throw IllegalArgumentException("Not a data size: ${it}. Example of correct values: '1024', '1kb', '5mb'")
+            val multiplier = m.groups[2]?.let {
+                when (it.value) {
+                    "k", "kb" -> 1024
+                    "m", "mb" -> 1024 * 1024
+                    else -> 1
+                }
+            } ?: 1
+            val base = m.groups[1]!!.value.toInt()
+            base * multiplier
+        }
+    }
+
     // ----
 
     fun getBlockchain(id: String): Chain {

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)
@@ -199,8 +195,9 @@ open class ConfiguredUpstreams(
         val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint ->
             val wsApi = EthereumWsFactory(
                     endpoint.url,
-                    endpoint.origin ?: URI("http://localhost")
+                    endpoint.origin ?: URI("http://localhost"),
             )
+            wsApi.config = endpoint
             endpoint.basicAuth?.let { auth ->
                 wsApi.basicAuth = auth
             }
@@ -209,13 +206,29 @@ open class ConfiguredUpstreams(
         }
 
         log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-        val ethereumUpstream = EthereumRpcUpstream(
+        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))
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt

@@ -61,7 +61,7 @@ abstract class DefaultUpstream(
         return status.get().status
     }
 
-    fun setStatus(avail: UpstreamAvailability) {
+    open fun setStatus(avail: UpstreamAvailability) {
         status.updateAndGet { curr ->
             Status(curr.lag, avail, statusByLag(curr.lag, avail))
         }

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

@@ -15,9 +15,48 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.AbstractHead
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.hex.HexQuantity
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
 
 open class DefaultEthereumHead : Head, AbstractHead() {
 
+    companion object {
+        private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
+    }
+
+    fun getLatestBlock(api: Reader<JsonRpcRequest, JsonRpcResponse>): Mono<BlockContainer> {
+        return api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
+                .subscribeOn(EthereumRpcHead.scheduler)
+                .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
+                .flatMap {
+                    if (it.error != null) {
+                        Mono.error(it.error.asException(null))
+                    } else {
+                        val value = it.getResultAsProcessedString()
+                        Mono.just(HexQuantity.from(value))
+                    }
+                }
+                .flatMap {
+                    //fetching by Block Height here, critical to use the same upstream as in previous call,
+                    //b/c different upstreams may have different blocks on the same height
+                    api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
+                            .subscribeOn(EthereumRpcHead.scheduler)
+                            .timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
+                }
+                .map {
+                    BlockContainer.fromEthereumJson(it.getResult())
+                }
+                .onErrorResume { err ->
+                    log.debug("Failed to fetch latest block: ${err.message}")
+                    Mono.empty()
+                }
+    }
 }

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

@@ -49,30 +49,7 @@ class EthereumRpcHead(
         val base = Flux.interval(interval)
                 .publishOn(scheduler)
                 .flatMap {
-                    api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
+                    getLatestBlock(api)
-                            .subscribeOn(scheduler)
-                            .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
-                            .flatMap {
-                                if (it.error != null) {
-                                    Mono.error(it.error.asException(null))
-                                } else {
-                                    val value = it.getResultAsProcessedString()
-                                    Mono.just(HexQuantity.from(value))
-                                }
-                            }
-                }
-                .flatMap {
-                    //fetching by Block Height here, critical to use same upstream,
-                    //different upstreams may have different blocks on the same height
-                    api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
-                            .subscribeOn(scheduler)
-                            .timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
-                }
-                .map {
-                    BlockContainer.fromEthereumJson(it.getResult())
-                }
-                .onErrorContinue { err, _ ->
-                    log.debug("RPC error ${err.message}")
                 }
         refreshSubscription = super.follow(base)
     }

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,8 @@ open class EthereumRpcUpstream(
 
     open fun createHead(): Head {
         return if (ethereumWsFactory != null) {
-            val ws = ethereumWsFactory.create(this).apply {
+            // 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 {
@@ -108,14 +104,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/EthereumUpstreamValidator.kt

@@ -32,7 +32,7 @@ import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.concurrent.Executors
 
-class EthereumUpstreamValidator(
+open class EthereumUpstreamValidator(
         private val upstream: EthereumUpstream,
         private val options: UpstreamsConfig.Options
 ) {
@@ -43,7 +43,7 @@ class EthereumUpstreamValidator(
 
     private val objectMapper: ObjectMapper = Global.objectMapper
 
-    fun validate(): Mono<UpstreamAvailability> {
+    open fun validate(): Mono<UpstreamAvailability> {
         return upstream
                 .getApi()
                 .read(JsonRpcRequest("eth_syncing", listOf()))

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

@@ -20,28 +20,45 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.config.AuthConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 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.emeraldpay.etherjar.rpc.ws.SubscriptionJson
+import io.netty.buffer.ByteBuf
 import io.netty.buffer.ByteBufInputStream
+import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpHeaderNames
+import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
+import org.springframework.util.backoff.BackOff
+import org.springframework.util.backoff.BackOffExecution
+import org.springframework.util.backoff.ExponentialBackOff
+import org.springframework.util.backoff.FixedBackOff
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.Sinks
+import reactor.core.scheduler.Schedulers
 import reactor.netty.http.client.HttpClient
 import reactor.netty.http.client.WebsocketClientSpec
+import reactor.netty.http.websocket.WebsocketInbound
+import reactor.netty.http.websocket.WebsocketOutbound
 import reactor.retry.Repeat
-import java.io.InputStream
+import reactor.util.function.Tuples
 import java.net.URI
 import java.time.Duration
 import java.util.*
+import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+
 
 class EthereumWsFactory(
         private val uri: URI,
@@ -49,30 +66,79 @@ class EthereumWsFactory(
 ) {
 
     var basicAuth: AuthConfig.ClientBasicAuth? = null
+    var config: UpstreamsConfig.WsEndpoint? = null
 
-    fun create(upstream: EthereumUpstream): EthereumWs {
+    fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): EthereumWs {
-        return EthereumWs(uri, origin, upstream, basicAuth)
+        return EthereumWs(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { ws ->
+            config?.frameSize?.let {
+                ws.frameSize = it
+            }
+            config?.msgSize?.let {
+                ws.msgSizeLimit = it
+            }
+        }
     }
 
     class EthereumWs(
             private val uri: URI,
             private val origin: URI,
-            private val upstream: EthereumUpstream,
+            private val basicAuth: AuthConfig.ClientBasicAuth?,
-            private val basicAuth: AuthConfig.ClientBasicAuth?
+            private val rpcMetrics: RpcMetrics?,
+            private val upstream: DefaultUpstream?,
+            private val validator: EthereumUpstreamValidator?
     ) : AutoCloseable {
 
         companion object {
             private val log = LoggerFactory.getLogger(EthereumWs::class.java)
 
+            private const val IDS_START = 100
             private const val START_REQUEST = "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}"
+
+            // WebSocket Frame limit.
+            // Default is 65_536, but Geth responds with larger frames,
+            // and connection gets dropped with:
+            // > io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException: Max frame length of 65536 has been exceeded
+            // It's unclear what is a right limit here, but 5mb seems to be working (1mb wasn't always working)
+            private const val DEFAULT_FRAME_SIZE = 5 * 1024 * 1024
+
+            // The max size from multiple frames that may represent a single message
+            // Accept up to 15Mb messages, because Geth is using 15mb, though it's not clear what it limits
+            private const val DEFAULT_MSG_SIZE = 15 * 1024 * 1024
         }
 
-        private val topic = Sinks
+        var frameSize: Int = DEFAULT_FRAME_SIZE
+        var msgSizeLimit: Int = DEFAULT_MSG_SIZE
+
+        private var reconnectBackoff: BackOff = ExponentialBackOff().also {
+            it.initialInterval = Duration.ofMillis(100).toMillis()
+            it.maxInterval = Duration.ofMinutes(1).toMillis()
+        }
+        private var currentBackOff = reconnectBackoff.start()
+
+        private val parser = ResponseWSParser()
+
+        private val blocks = Sinks
                 .many()
                 .multicast()
                 .directBestEffort<BlockContainer>()
+        private var rpcSend = Sinks
+                .many()
+                .unicast()
+                .onBackpressureBuffer<JsonRpcRequest>()
+        private val rpcReceive = Sinks
+                .many()
+                .multicast()
+                .directBestEffort<JsonRpcResponse>()
+        private val sendIdSeq = AtomicInteger(IDS_START)
+        private val sendExecutor = Executors.newSingleThreadExecutor()
         private var keepConnection = true
         private var connection: Disposable? = null
+        private val reconnecting = AtomicBoolean(false)
+
+        fun setReconnectIntervalSeconds(value: Long) {
+            reconnectBackoff = FixedBackOff(value * 1000, FixedBackOff.UNLIMITED_ATTEMPTS)
+            currentBackOff = reconnectBackoff.start()
+        }
 
         fun connect() {
             if (keepConnection) {
@@ -81,29 +147,53 @@ class EthereumWsFactory(
         }
 
         private fun tryReconnectLater() {
+            if (!keepConnection) {
+                return
+            }
+            val alreadyReconnecting = reconnecting.getAndSet(true)
+            if (alreadyReconnecting) {
+                return
+            }
+            // rpcSend is already CANCELLED, since the subscription owned by the previous connection is gone
+            // so we need to create a new Sink. Emit Complete is probably useless, and just in case
+            rpcSend.tryEmitComplete()
+            rpcSend = Sinks
+                    .many()
+                    .unicast()
+                    .onBackpressureBuffer<JsonRpcRequest>()
+            val retryInterval = currentBackOff.nextBackOff()
+            if (retryInterval == BackOffExecution.STOP) {
+                log.warn("Reconnect backoff exhausted. Permanently closing the connection")
+                return
+            }
+            log.info("Reconnect to $uri in ${retryInterval}ms...")
             Global.control.schedule(
-                    { connectInternal() },
+                    {
-                    Defaults.retryConnection.seconds, TimeUnit.SECONDS)
+                        reconnecting.set(false)
+                        connectInternal()
+                    },
+                    retryInterval, TimeUnit.MILLISECONDS)
         }
 
         private fun connectInternal() {
             log.info("Connecting to WebSocket: $uri")
             connection?.dispose()
-            connection = null
-
-            val subscriptionId = AtomicReference<String>("NOTSET")
-
-            val objectMapper = Global.objectMapper
             connection = HttpClient.create()
+                    .doOnDisconnected {
+                        log.info("Disconnected from $uri")
+                        // mark upstream as UNAVAIL
+                        upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
+                        if (keepConnection) {
+                            tryReconnectLater()
+                        }
+                    }
                     .doOnError(
                             { _, t ->
                                 log.warn("Failed to connect to $uri. Error: ${t.message}")
                                 // going to try to reconnect later
                                 tryReconnectLater()
                             },
-                            { _, _ ->
+                            { _, _ -> }
-
-                            }
                     )
                     .headers { headers ->
                         headers.add(HttpHeaderNames.ORIGIN, origin)
@@ -114,71 +204,110 @@ class EthereumWsFactory(
                         }
                     }
                     .let {
-                        if (uri.scheme == "wss") {
+                        if (uri.scheme == "wss") it.secure() else it
-                            it.secure()
-                        } else {
-                            it
-                        }
                     }
                     .websocket(
                             WebsocketClientSpec.builder()
                                     .handlePing(true)
                                     .compress(false)
+                                    .maxFramePayloadLength(frameSize)
                                     .build()
                     )
                     .uri(uri)
                     .handle { inbound, outbound ->
-                        val consumer = inbound.aggregateFrames()
+                        handle(inbound, outbound)
-                                .aggregateFrames(8 * 65_536)
+                    }
+                    .onErrorResume { t ->
+                        log.debug("Dropping WS connection to $uri. Error: ${t.message}")
+                        Mono.empty<Void>()
+                    }
+                    .subscribe()
+        }
+
+        fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
+            //restart backoff after connection
+            currentBackOff = reconnectBackoff.start()
+
+            //validate the connection, it can also be UNAVAIL if market as such after disconnect
+            validator?.validate()
+
+            val consumer = inbound
+                    .aggregateFrames(msgSizeLimit)
                     .receiveFrames()
+                    .map { ByteBufInputStream(it.content()).readAllBytes() }
                     .flatMap {
-                                    val msg: SubscriptionJson = objectMapper.readerFor(SubscriptionJson::class.java)
+                        try {
-                                            .readValue(ByteBufInputStream(it.content()) as InputStream)
+                            val msg = parser.parse(it)
-                                    when {
+                            if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) {
-                                        msg.error != null -> {
+                                onSubscription(msg)
-                                            Mono.error(IllegalStateException("Received error from WS upstream"))
+                            } else {
-                                        }
+                                onRpc(msg)
-                                        msg.subscription == subscriptionId.get() -> {
-                                            onNewBlock(msg.blockResult)
-                                            Mono.empty<Int>()
-                                        }
-                                        msg.subscription == null -> {
-                                            // received ID for subscription
-                                            subscriptionId.set(msg.result.asText())
-                                            log.debug("Connected to $uri")
-                                            Mono.empty<Int>()
-                                        }
-                                        else -> {
-                                            Mono.error(IllegalStateException("Unknown message received: ${msg.subscription}"))
                             }
+                        } catch (t: Throwable) {
+                            log.warn("Failed to process WS message. ${t.javaClass}: ${t.message}")
+                            Mono.empty()
                         }
                     }
                     .onErrorResume { t ->
-                                    log.warn("Connection dropped to $uri. Error: ${t.message}")
+                        log.warn("Connection dropped to $uri. Error: ${t.message}", t)
                         // going to try to reconnect later
                         tryReconnectLater()
                         // completes current outbound flow
                         Mono.empty()
                     }
 
-
+            val start = Mono.just(START_REQUEST).map {
-                        outbound.sendString(Mono.just(START_REQUEST)
+                Unpooled.wrappedBuffer(it.toByteArray())
-                                .doOnError { log.warn("Failed to start WS subscription. ${it.javaClass}: ${it.message}") })
-                                .then(consumer.then())
             }
-                    .doOnError {
+            val calls = rpcSend
-                        println(it)
+                    .asFlux()
-                    }
+                    .map {
-                    .subscribe()
+                        Unpooled.wrappedBuffer(Global.objectMapper.writeValueAsBytes(it))
                     }
 
-        fun onNewBlock(block: BlockJson<TransactionRefJson>) {
+            return outbound.send(
-            // WS returns incomplete blocks, i.e. without some fields, so need to fetch full block data
+                    Flux.merge(
-            if (block.difficulty == null || block.transactions == null) {
+                            start,
+                            calls.subscribeOn(Schedulers.boundedElastic()),
+                            consumer.then(Mono.empty<ByteBuf>()).subscribeOn(Schedulers.boundedElastic())
+                    )
+            )
+        }
+
+        fun onRpc(msg: ResponseWSParser.WsResponse): Mono<Void> {
+            return if (msg.id.isNumber()) {
+                val resp = JsonRpcResponse(
+                        msg.value, msg.error, msg.id
+                )
+                Mono.fromCallable {
+                    val status = rpcReceive.tryEmitNext(resp)
+                    if (status.isFailure) {
+                        log.warn("Failed to proceed with a RPC message: $status")
+                    }
+                }.then()
+            } else {
+                //it's a response to the newHeads subscription, just ignore it
+                Mono.empty<Void>()
+            }
+        }
+
+        fun onSubscription(msg: ResponseWSParser.WsResponse): Mono<Void> {
+            if (msg.error != null) {
+                return Mono.error(IllegalStateException("Received error from WS upstream: ${msg.error.message}"))
+            }
+            // we always expect an answer to the `newHeads`, since we are not initiating any other subscriptions
+            return Mono.fromCallable {
+                Global.objectMapper.readValue(msg.value, BlockJson::class.java) as BlockJson<TransactionRefJson>
+            }.flatMap { onNewHeads(it) }.then()
+        }
+
+        fun onNewHeads(block: BlockJson<TransactionRefJson>): Mono<Void> {
+            // newHeads returns incomplete blocks, i.e. without some fields and without transaction hashes,
+            // so we need to fetch the full block data
+            return if (block.difficulty == null || block.transactions == null) {
                 Mono.just(block.hash)
                         .flatMap { hash ->
-                            upstream.getApi()
+                            call(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false)))
-                                    .read(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false)))
                                     .flatMap { resp ->
                                         if (resp.isNull()) {
                                             Mono.error(SilentException("Received null for block $hash"))
@@ -188,6 +317,8 @@ class EthereumWsFactory(
                                     }
                                     .flatMap(JsonRpcResponse::requireResult)
                                     .map { BlockContainer.fromEthereumJson(it) }
+                                    .subscribeOn(Schedulers.boundedElastic())
+                                    .timeout(Defaults.timeoutInternal, Mono.empty())
                         }.repeatWhenEmpty { n ->
                             Repeat.times<Any>(5)
                                     .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500))
@@ -195,17 +326,60 @@ class EthereumWsFactory(
                         }
                         .timeout(Defaults.timeout, Mono.empty())
                         .onErrorResume { Mono.empty() }
-                        .subscribe {
+                        .doOnNext {
-                            topic.tryEmitNext(it)
+                            blocks.tryEmitNext(it)
                         }
-
+                        .then()
             } else {
-                topic.tryEmitNext(BlockContainer.from(block))
+                Mono.fromCallable {
+                    blocks.tryEmitNext(BlockContainer.from(block))
+                }.then()
             }
         }
 
-        fun getFlux(): Flux<BlockContainer> {
+        fun call(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse> {
-            return this.topic.asFlux()
+            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, startTime)
+            }.flatMap { request ->
+                waitForResponse(request.t1, request.t2, request.t3)
+            }
+        }
+
+        fun sendRpc(request: JsonRpcRequest) {
+            // submit to upstream in a separate thread, to free current thread (needs for subscription, etc)
+            sendExecutor.execute {
+                val result = rpcSend.tryEmitNext(request)
+                if (result.isFailure) {
+                    log.warn("Failed to send RPC request: $result")
+                }
+            }
+        }
+
+        fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
+            val expectedId = request.id.toLong()
+            return Mono.just(request)
+                    .flatMap {
+                        Flux.from(rpcReceive.asFlux())
+                                .doOnSubscribe { sendRpc(request) }
+                                .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)) }
+                    }
+        }
+
+        fun getBlocksFlux(): Flux<BlockContainer> {
+            return this.blocks.asFlux()
         }
 
         override fun close() {
@@ -216,5 +390,4 @@ class EthereumWsFactory(
     }
 
 
-
 }

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

@@ -16,9 +16,11 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
+import reactor.core.publisher.Flux
 
 class EthereumWsHead(
         private val ws: EthereumWsFactory.EthereumWs
@@ -34,7 +36,12 @@ class EthereumWsHead(
 
     override fun start() {
         this.subscription?.dispose()
-        this.subscription = super.follow(ws.getFlux())
+        val heads = Flux.merge(
+                // get the current block, not just wait for the next update
+                getLatestBlock(JsonRpcWsClient(ws)),
+                ws.getBlocksFlux()
+        )
+        this.subscription = super.follow(heads)
     }
 
     override fun stop() {

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

@@ -0,0 +1,122 @@
+/**
+ * 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
+    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.err")
+                        .description("Errors received on request through WebSocket JSON RPC connection")
+                        .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()
+
+        log.debug("Start validation for upstream ${this.getId()}")
+        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/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt

@@ -123,8 +123,8 @@ class NativeCallRouter(
         }
     }
 
-    fun getBlockByNumber(params: List<Any>): Mono<ByteArray>? {
+    fun getBlockByNumber(params: List<Any?>): Mono<ByteArray>? {
-        if (params.size != 2) {
+        if (params.size != 2 || params[0] == null || params[1] == null) {
             throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
         }
         val number: Long
@@ -155,7 +155,7 @@ class NativeCallRouter(
                 }
             }
         } catch (e: IllegalArgumentException) {
-            throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number")
+            throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be a block number")
         }
         val withTx = params[1].toString().toBoolean()
         var block = reader.blocksByHeightAsCont()

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt

@@ -52,5 +52,9 @@ class JsonRpcError(val code: Int, val message: String, val details: Any?) {
         return result
     }
 
+    override fun toString(): String {
+        return "JsonRpcError(code=$code, message='$message', details=$details)"
+    }
+
 
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt

@@ -19,8 +19,6 @@ import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
-import io.micrometer.core.instrument.Counter
-import io.micrometer.core.instrument.Timer
 import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpHeaders
@@ -50,7 +48,7 @@ class JsonRpcHttpClient(
         private val log = LoggerFactory.getLogger(JsonRpcHttpClient::class.java)
     }
 
-    private val parser = JsonRpcParser()
+    private val parser = ResponseRpcParser()
     private val httpClient: HttpClient
 
     init {

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt

@@ -1,111 +0,0 @@
-/**
- * Copyright (c) 2020 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.rpcclient
-
-import com.fasterxml.jackson.core.JsonFactory
-import com.fasterxml.jackson.core.JsonParseException
-import com.fasterxml.jackson.core.JsonParser
-import com.fasterxml.jackson.core.JsonToken
-import io.emeraldpay.dshackle.Global
-import io.emeraldpay.etherjar.rpc.RpcResponseError
-import org.slf4j.LoggerFactory
-
-class JsonRpcParser() {
-
-    companion object {
-        private val log = LoggerFactory.getLogger(JsonRpcParser::class.java)
-    }
-
-    private val jsonFactory = JsonFactory()
-
-    fun parse(json: ByteArray): JsonRpcResponse {
-        try {
-            val parser: JsonParser = jsonFactory.createParser(json)
-            parser.nextToken()
-            if (parser.currentToken != JsonToken.START_OBJECT) {
-                return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON"))
-            }
-            var nullResponse: JsonRpcResponse? = null
-            while (parser.nextToken() != JsonToken.END_OBJECT) {
-                val field = parser.currentName
-                if (field == "jsonrpc" || field == "id") {
-                    if (!parser.nextToken().isScalarValue) {
-                        return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id or jsonrpc value)"))
-                    }
-                    // just skip the field
-                } else if (field == "result") {
-                    val value = parser.nextToken()
-                    val start = parser.tokenLocation
-                    if (value.isScalarValue) {
-                        val text = parser.text
-                        if (value == JsonToken.VALUE_STRING) {
-                            return JsonRpcResponse(("\"" + text + "\"").toByteArray(), null)
-                        } else if (value == JsonToken.VALUE_NULL) {
-                            //if null we should check if error is present
-                            nullResponse = JsonRpcResponse(text.toByteArray(), null)
-                        } else {
-                            return JsonRpcResponse(text.toByteArray(), null)
-                        }
-                    } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) {
-                        parser.skipChildren()
-                        val end = parser.currentLocation.byteOffset.toInt()
-                        val copy = ByteArray((end - start.byteOffset).toInt())
-                        System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size)
-                        return JsonRpcResponse(copy, null)
-                    }
-                } else if (field == "error") {
-                    val err = readError(parser)
-                    if (err != null) {
-                        return JsonRpcResponse(null, err)
-                    }
-                }
-            }
-            if (nullResponse != null) {
-                return nullResponse
-            }
-        } catch (e: JsonParseException) {
-            log.warn("Failed to parse JSON from upstream: ${e.message}")
-        }
-        return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure"))
-    }
-
-    fun readError(parser: JsonParser): JsonRpcError? {
-        var code = 0
-        var message = ""
-        var details: Any? = null
-
-        while (parser.nextToken() != JsonToken.END_OBJECT) {
-            if (parser.currentToken() == JsonToken.VALUE_NULL) {
-                // error is just null
-                return null
-            }
-            val field = parser.currentName()
-            if (field == "code" && parser.currentToken == JsonToken.VALUE_NUMBER_INT) {
-                code = parser.intValue
-            } else if (field == "message" && parser.currentToken == JsonToken.VALUE_STRING) {
-                message = parser.valueAsString
-            } else if (field == "data") {
-                when (val value = parser.nextToken()) {
-                    JsonToken.VALUE_NULL -> details = null
-                    JsonToken.VALUE_STRING -> details = parser.valueAsString
-                    JsonToken.START_OBJECT -> details = Global.objectMapper.readValue(parser, java.util.Map::class.java)
-                    else -> log.warn("Unsupported error data type $value")
-                }
-            }
-        }
-        return JsonRpcError(code, message, details)
-    }
-}

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt

@@ -15,40 +15,55 @@
  */
 package io.emeraldpay.dshackle.upstream.rpcclient
 
+import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.databind.DeserializationContext
+import com.fasterxml.jackson.databind.JsonDeserializer
+import com.fasterxml.jackson.databind.JsonNode
 import io.emeraldpay.dshackle.Global
 
-class JsonRpcRequest(
+data class JsonRpcRequest(
         val method: String,
-        val params: List<Any>
+        val params: List<Any?>,
+        val id: Int
 ) {
 
+    constructor(method: String, params: List<Any?>) : this(method, params, 1)
+
     fun toJson(): ByteArray {
         val json = mapOf(
                 "jsonrpc" to "2.0",
-                "id" to 1,
+                "id" to id,
                 "method" to method,
                 "params" to params
         )
         return Global.objectMapper.writeValueAsBytes(json)
     }
 
-    override fun equals(other: Any?): Boolean {
-        if (this === other) return true
-        if (other !is JsonRpcRequest) return false
-
-        if (method != other.method) return false
-        if (params != other.params) return false
-
-        return true
-    }
-
-    override fun hashCode(): Int {
-        var result = method.hashCode()
-        result = 31 * result + params.hashCode()
-        return result
-    }
-
     override fun toString(): String {
         return String(this.toJson())
     }
+
+    class Deserializer : JsonDeserializer<JsonRpcRequest>() {
+
+        override fun deserialize(p: JsonParser, ctxt: DeserializationContext): JsonRpcRequest {
+            val node: JsonNode = p.readValueAsTree()
+            val id = node.get("id").intValue()
+            val method = node.get("method").textValue()
+            val params = node.get("params").map {
+                if (it.isNumber) {
+                    it.asInt()
+                } else if (it.isTextual) {
+                    it.textValue()
+                } else if (it.isBoolean) {
+                    it.booleanValue()
+                } else if (it.isNull) {
+                    null
+                } else {
+                    throw IllegalStateException("Unsupported param type: ${it.asToken()}")
+                }
+            }
+            return JsonRpcRequest(method, params, id)
+        }
+
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt

@@ -106,6 +106,10 @@ class JsonRpcResponse(
         }
     }
 
+    fun copyWithId(id: Id): JsonRpcResponse {
+        return JsonRpcResponse(result, error, id)
+    }
+
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
         if (other !is JsonRpcResponse) return false
@@ -177,6 +181,10 @@ class JsonRpcResponse(
         override fun hashCode(): Int {
             return id.hashCode()
         }
+
+        override fun toString(): String {
+            return id.toString()
+        }
     }
 
     class StringId(val id: String) : Id {
@@ -205,6 +213,9 @@ class JsonRpcResponse(
             return id.hashCode()
         }
 
+        override fun toString(): String {
+            return id
+        }
     }
 
     class ResponseJsonSerializer : JsonSerializer<JsonRpcResponse>() {

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt

@@ -0,0 +1,30 @@
+/**
+ * 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.rpcclient
+
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
+import reactor.core.publisher.Mono
+
+class JsonRpcWsClient(
+        private val ws: EthereumWsFactory.EthereumWs
+) : Reader<JsonRpcRequest, JsonRpcResponse> {
+
+    override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+        return ws.call(key)
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt

@@ -0,0 +1,182 @@
+/**
+ * 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.rpcclient
+
+import com.fasterxml.jackson.core.JsonFactory
+import com.fasterxml.jackson.core.JsonParseException
+import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.core.JsonToken
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.etherjar.rpc.RpcResponseError
+import org.slf4j.LoggerFactory
+import java.io.IOException
+
+abstract class ResponseParser<T> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ResponseParser::class.java)
+    }
+
+    private val jsonFactory = JsonFactory()
+
+    abstract fun build(state: Preparsed): T
+
+    fun parse(json: ByteArray): T {
+        return build(parseInternal(json))
+    }
+
+    private fun parseInternal(json: ByteArray): Preparsed {
+        var state = Preparsed()
+        try {
+            val parser: JsonParser = jsonFactory.createParser(json)
+            parser.nextToken()
+            if (parser.currentToken != JsonToken.START_OBJECT) {
+                return Preparsed(error = JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON: not an Object"))
+            }
+            while (parser.nextToken() != JsonToken.END_OBJECT) {
+                val field = parser.currentName
+                state = process(parser, json, field, state)
+            }
+        } catch (e: JsonParseException) {
+            log.warn("Failed to parse JSON from upstream: ${e.message}")
+        }
+        if (state.isReady) {
+            return state
+        }
+        return Preparsed(error = JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure: never finalized"))
+    }
+
+    open fun process(parser: JsonParser, json: ByteArray, field: String, state: Preparsed): Preparsed {
+        if (field == "jsonrpc") {
+            if (!parser.nextToken().isScalarValue) {
+                return state.copy(error = JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (jsonrpc value)"))
+            }
+            // just skip the field
+            return state
+        } else if (field == "id") {
+            return state.copy(id = readId(parser))
+        } else if (field == "result") {
+            val result = readResult(json, parser)
+            return if (result == null) {
+                //if result is null we should check if an error is also present, and if it's set then return only the error
+                state.copy(nullResult = true)
+            } else {
+                state.copy(result = result)
+            }
+        } else if (field == "error") {
+            val err = readError(parser)
+            if (err != null) {
+                return state.copy(error = err)
+            }
+        }
+        return state
+    }
+
+    private fun readId(parser: JsonParser): JsonRpcResponse.Id {
+        if (parser.currentToken() == JsonToken.FIELD_NAME) {
+            parser.nextToken()
+        }
+        return if (parser.currentToken() == JsonToken.VALUE_NUMBER_INT) {
+            JsonRpcResponse.NumberId(parser.intValue)
+        } else if (parser.currentToken() == JsonToken.VALUE_STRING) {
+            JsonRpcResponse.StringId(parser.text)
+        } else {
+            throw IllegalStateException("Not a string or number: ${parser.currentToken()}")
+        }
+    }
+
+    @Throws(IOException::class)
+    private fun readNumber(parser: JsonParser): Int {
+        if (parser.currentToken() != JsonToken.VALUE_NUMBER_INT) {
+            parser.nextToken()
+        }
+        if (!parser.currentToken().isNumeric) {
+            throw IllegalStateException("Not a number: ${parser.currentToken.name}")
+        }
+        return parser.intValue
+    }
+
+    fun readResult(json: ByteArray, parser: JsonParser): ByteArray? {
+        val value = parser.nextToken()
+        val start = parser.tokenLocation
+        if (value.isScalarValue) {
+            val text = parser.text
+            return if (value == JsonToken.VALUE_STRING) {
+                ("\"" + text + "\"").toByteArray()
+            } else if (value == JsonToken.VALUE_NULL) {
+                null
+            } else {
+                text.toByteArray()
+            }
+        } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) {
+            parser.skipChildren()
+            val end = parser.currentLocation.byteOffset.toInt()
+            val copy = ByteArray((end - start.byteOffset).toInt())
+            System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size)
+            return copy
+        } else {
+            throw IllegalStateException("Invalid JSON structure, cannot read result from ${value.name}")
+        }
+    }
+
+    fun readError(parser: JsonParser): JsonRpcError? {
+        var code = 0
+        var message = ""
+        var details: Any? = null
+
+        while (parser.nextToken() != JsonToken.END_OBJECT) {
+            if (parser.currentToken() == JsonToken.VALUE_NULL) {
+                // error is just null
+                return null
+            }
+            val field = parser.currentName()
+            if (field == "code" && parser.currentToken == JsonToken.VALUE_NUMBER_INT) {
+                code = parser.intValue
+            } else if (field == "message" && parser.currentToken == JsonToken.VALUE_STRING) {
+                message = parser.valueAsString
+            } else if (field == "data") {
+                when (val value = parser.nextToken()) {
+                    JsonToken.VALUE_NULL -> details = null
+                    JsonToken.VALUE_STRING -> details = parser.valueAsString
+                    JsonToken.START_OBJECT -> details = Global.objectMapper.readValue(parser, java.util.Map::class.java)
+                    else -> log.warn("Unsupported error data type $value")
+                }
+            }
+        }
+        return JsonRpcError(code, message, details)
+    }
+
+    data class Preparsed(
+            val id: JsonRpcResponse.Id? = null,
+            val result: ByteArray? = null,
+            val nullResult: Boolean = false,
+            val error: JsonRpcError? = null,
+            val subMethod: String? = null,
+            val subId: String? = null
+    ) {
+
+        private val isResultSet = result != null || nullResult
+
+        val isRpcReady: Boolean = id != null &&
+                (error != null || isResultSet)
+
+        val isSubReady: Boolean = subId != null &&
+                isResultSet
+
+        val isReady: Boolean = isRpcReady || isSubReady
+
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt

@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2020 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.rpcclient
+
+import org.slf4j.LoggerFactory
+
+open class ResponseRpcParser() : ResponseParser<JsonRpcResponse>() {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ResponseRpcParser::class.java)
+    }
+
+    override fun build(state: Preparsed): JsonRpcResponse {
+        if (state.error != null) {
+            return JsonRpcResponse(null, state.error, state.id ?: JsonRpcResponse.Id.from(-1))
+        }
+        if (state.nullResult) {
+            return JsonRpcResponse("null".toByteArray(), null, state.id ?: JsonRpcResponse.Id.from(-1))
+        }
+        return JsonRpcResponse(state.result, null, state.id ?: JsonRpcResponse.Id.from(-1))
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt

@@ -0,0 +1,110 @@
+/**
+ * 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.rpcclient
+
+import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.core.JsonToken
+import org.slf4j.LoggerFactory
+import java.io.IOException
+
+
+class ResponseWSParser : ResponseParser<ResponseWSParser.WsResponse>() {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ResponseWSParser::class.java)
+        private val NULL_RESULT = "null".toByteArray()
+    }
+
+    override fun build(state: Preparsed): WsResponse {
+        if (state.isRpcReady) {
+            return WsResponse(
+                    Type.RPC,
+                    state.id!!,
+                    if (state.nullResult) NULL_RESULT else state.result,
+                    state.error
+            )
+        }
+        if (state.isSubReady) {
+            return WsResponse(
+                    Type.SUBSCRIPTION,
+                    JsonRpcResponse.Id.from(state.subId!!),
+                    if (state.nullResult) NULL_RESULT else state.result,
+                    state.error
+            )
+        }
+        throw IllegalStateException("State is not ready")
+    }
+
+    override fun process(parser: JsonParser, json: ByteArray, field: String, state: Preparsed): Preparsed {
+        if ("method" == field) {
+            parser.nextToken()
+            val method = parser.getValueAsString()
+            return state.copy(subMethod = method)
+        }
+        if ("params" == field) {
+            // example:
+            // newHeads
+            // {
+            //  "jsonrpc": "2.0",
+            //  "method": "eth_subscription",
+            //  "params": {
+            //    "result": {
+            //      "difficulty": ......
+            //    },
+            //    "subscription": "...."
+            //  }
+            //}
+            return decodeSubscription(parser, json, state)
+        }
+        return super.process(parser, json, field, state)
+    }
+
+    @Throws(IOException::class)
+    private fun decodeString(parser: JsonParser): String {
+        if (parser.currentToken() != JsonToken.VALUE_STRING) {
+            parser.nextToken()
+        }
+        check(parser.currentToken().isScalarValue) { "Id is not a string" }
+        return parser.valueAsString
+    }
+
+    @Throws(IOException::class)
+    protected fun decodeSubscription(parser: JsonParser, json: ByteArray, stateOriginal: Preparsed): Preparsed {
+        var state = stateOriginal
+        while (parser.nextToken() != JsonToken.END_OBJECT) {
+            checkNotNull(parser.currentToken()) { "JSON finished before data received" }
+            val field = parser.currentName()
+            if ("subscription" == field) {
+                state = state.copy(subId = decodeString(parser))
+            } else if ("result" == field) {
+                state = state.copy(result = readResult(json, parser))
+            }
+        }
+        return state
+    }
+
+    enum class Type {
+        SUBSCRIPTION, RPC
+    }
+
+    data class WsResponse(
+            val type: Type,
+            val id: JsonRpcResponse.Id,
+            val value: ByteArray?,
+            val error: JsonRpcError?
+    )
+
+}

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

@@ -74,6 +74,58 @@ 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 full defined websocket config"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("upstreams-ws-full.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"
+                }
+                ws.frameSize == 10 * 1024 * 1024
+                ws.msgSize == 25 * 1024 * 1024
+            }
+        }
+    }
+
     def "Parse bitcoin upstreams"() {
         setup:
         def config = this.class.getClassLoader().getResourceAsStream("upstreams-bitcoin.yaml")

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

@@ -0,0 +1,45 @@
+/**
+ * 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.config
+
+import org.yaml.snakeyaml.Yaml
+import org.yaml.snakeyaml.nodes.MappingNode
+import spock.lang.Specification
+
+class YamlConfigReaderSpec extends Specification {
+
+    def "reads bytes values"() {
+        setup:
+        def rdr = new Impl()
+        expect:
+        rdr.getValueAsBytes(asNode("test", input), "test") == exp
+        where:
+        input  | exp
+        "1024" | 1024
+        "1k"   | 1024
+        "1kb"  | 1024
+        "1K"   | 1024
+        "16kb" | 16 * 1024
+        "1M"   | 1024 * 1024
+        "4mb"  | 4 * 1024 * 1024
+    }
+
+    private MappingNode asNode(String key, String value) {
+        return new Yaml().compose(new StringReader("$key: $value")) as MappingNode
+    }
+
+    class Impl extends YamlConfigReader {}
+}

src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy

@@ -27,13 +27,34 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.grpc.stub.StreamObserver
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import io.emeraldpay.etherjar.rpc.json.ResponseJson
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+import io.netty.buffer.ByteBufInputStream
+import io.netty.buffer.Unpooled
+import io.netty.handler.codec.http.HttpHeaders
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
+import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus
+import io.netty.handler.codec.http.websocketx.WebSocketFrame
 import org.jetbrains.annotations.NotNull
+import org.reactivestreams.Publisher
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
+import reactor.netty.ByteBufFlux
+import reactor.netty.Connection
+import reactor.netty.NettyInbound
+import reactor.netty.NettyOutbound
+import reactor.netty.http.websocket.WebsocketInbound
+import reactor.netty.http.websocket.WebsocketOutbound
+import reactor.util.annotation.Nullable
 
 import java.time.Duration
 import java.util.concurrent.Callable
+import java.util.function.BiFunction
+import java.util.function.Consumer
+import java.util.function.Predicate
 
 class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
 
@@ -57,7 +78,7 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
     }
 
     @Override
-    Mono<JsonRpcResponse> read(JsonRpcRequest request) {
+    Mono<JsonRpcResponse> read(JsonRpcRequest request, boolean required = true) {
         Callable<JsonRpcResponse> call = {
             def predefined = predefined.find { it.isSame(request.method, request.params) }
             byte[] result = null
@@ -65,7 +86,7 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
             if (predefined != null) {
                 if (predefined.exception != null) {
                     predefined.onCalled()
-                    predefined.print()
+                    predefined.print(request.id)
                     throw predefined.exception
                 }
                 if (predefined.result instanceof RpcResponseError) {
@@ -77,12 +98,15 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
                     result = objectMapper.writeValueAsBytes(predefined.result)
                 }
                 predefined.onCalled()
-                predefined.print()
+                predefined.print(request.id)
             } else {
                 log.error("Method ${request.method} with ${request.params} is not mocked")
+                if (!required) {
+                    return null
+                }
                 error = new JsonRpcError(-32601, "Method ${request.method} with ${request.params} is not mocked")
             }
-            return new JsonRpcResponse(result, error)
+            return new JsonRpcResponse(result, error, JsonRpcResponse.Id.from(request.id))
         } as Callable<JsonRpcResponse>
         return Mono.fromCallable(call)
     }
@@ -104,6 +128,10 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
         responseObserver.onCompleted()
     }
 
+    WebsocketApi asWebsocket() {
+        return new WebsocketApi(this)
+    }
+
     class PredefinedResponse {
         String method
         List params
@@ -132,8 +160,202 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
             }
         }
 
-        void print() {
+        void print(int id) {
-            println "Execute API: $method ${params ? params : '_'} >> $result"
+            println "Execute API: $id $method ${params ? params : '_'} >> $result"
+        }
+    }
+
+    class WebsocketApi {
+        private final EthereumApiMock api
+
+        private Sinks.Many<JsonRpcResponse> responses = Sinks
+                .many()
+                .unicast()
+                .onBackpressureBuffer()
+        private Sinks.Many<String> jsonResponses = Sinks
+                .many()
+                .unicast()
+                .onBackpressureBuffer()
+        private WebsocketOutboundMock outbound
+        private WebsocketInboundMock inbound
+
+        WebsocketApi(EthereumApiMock api) {
+            this.api = api
+            outbound = new WebsocketOutboundMock(api, responses)
+            inbound = new WebsocketInboundMock(responses.asFlux(), jsonResponses.asFlux())
+        }
+
+        boolean send(String json) {
+            jsonResponses.tryEmitNext(json).success
+        }
+
+        WebsocketOutbound getOutbound() {
+            return outbound
+        }
+
+        WebsocketInbound getInbound() {
+            return inbound
+        }
+    }
+
+    class WebsocketInboundMock implements WebsocketInbound {
+
+        private final Flux<JsonRpcResponse> responses
+        private final Flux<String> jsonResponses
+
+        WebsocketInboundMock(Flux<JsonRpcResponse> responses, Flux<String> jsonResponses) {
+            this.responses = responses
+            this.jsonResponses = jsonResponses
+        }
+
+        @Override
+        String selectedSubprotocol() {
+            throw new UnsupportedOperationException()
+        }
+
+        @Override
+        HttpHeaders headers() {
+            throw new UnsupportedOperationException()
+        }
+
+        @Override
+        Mono<WebSocketCloseStatus> receiveCloseStatus() {
+            return Mono.empty()
+        }
+
+        @Override
+        ByteBufFlux receive() {
+            throw new UnsupportedOperationException()
+        }
+
+        @Override
+        Flux<?> receiveObject() {
+            throw new UnsupportedOperationException()
+        }
+
+        @Override
+        NettyInbound withConnection(Consumer<? super Connection> withConnection) {
+            return this
+        }
+
+        @Override
+        Flux<WebSocketFrame> receiveFrames() {
+            return Flux.merge(
+                    jsonResponses,
+                    responses.map {
+                        Global.objectMapper.writeValueAsString(it)
+                    })
+                    .map {
+                        println("WS server->client msg: $it")
+                        new TextWebSocketFrame(it)
+                    }
+                    .doOnError { t ->
+                        t.printStackTrace()
+                    }
+        }
+    }
+
+    class WebsocketOutboundMock implements WebsocketOutbound {
+
+        private final EthereumApiMock api
+        private final Sinks.Many<JsonRpcResponse> responses
+
+        WebsocketOutboundMock(EthereumApiMock api, Sinks.Many<JsonRpcResponse> responses) {
+            this.api = api
+            this.responses = responses
+        }
+
+        @Override
+        String selectedSubprotocol() {
+            throw new UnsupportedOperationException()
+        }
+
+        @Override
+        ByteBufAllocator alloc() {
+            throw new UnsupportedOperationException()
+        }
+
+        private void handle(Publisher<ByteBuf> dataStream) {
+            Flux.from(dataStream)
+                    .map { it ->
+                        Global.objectMapper.readValue(new ByteBufInputStream(it), JsonRpcRequest)
+                    }
+                    .flatMap { JsonRpcRequest request ->
+                        api.read(request, false)
+                    }
+                    .doOnNext {
+                        def status = responses.tryEmitNext(it)
+                        if (status.isFailure()) {
+                            println("Failed to send through mock: $status")
+                        }
+                    }
+                    .subscribe()
+        }
+
+        @Override
+        NettyOutbound send(Publisher<? extends ByteBuf> dataStream) {
+            handle(dataStream)
+            return this
+        }
+
+        @Override
+        NettyOutbound send(Publisher<? extends ByteBuf> dataStream, Predicate<ByteBuf> predicate) {
+            handle(dataStream)
+            return this
+        }
+
+        @Override
+        NettyOutbound sendObject(Publisher<?> dataStream, Predicate<Object> predicate) {
+            def msgs = Flux.from(dataStream)
+                    .cast(TextWebSocketFrame)
+                    .map {
+                        Unpooled.wrappedBuffer(it.text().bytes)
+                    }
+            handle(msgs)
+            return this
+        }
+
+        @Override
+        NettyOutbound sendObject(Object message) {
+            return this
+        }
+
+        @Override
+        def <S> NettyOutbound sendUsing(Callable<? extends S> sourceInput, BiFunction<? super Connection, ? super S, ?> mappedInput, Consumer<? super S> sourceCleanup) {
+            return this
+        }
+
+        @Override
+        NettyOutbound withConnection(Consumer<? super Connection> withConnection) {
+            return this
+        }
+
+        @Override
+        Mono<Void> sendClose() {
+            return Mono.fromCallable {
+                responses.tryEmitComplete()
+            }.then()
+        }
+
+        @Override
+        Mono<Void> sendClose(int rsv) {
+            return Mono.fromCallable {
+                responses.tryEmitComplete()
+            }.then()
+        }
+
+        @Override
+        Mono<Void> sendClose(int statusCode, @Nullable String reasonText) {
+            return Mono.fromCallable {
+                responses.tryEmitComplete()
+            }.then()
+        }
+
+        @Override
+        Mono<Void> sendClose(int rsv, int statusCode, @Nullable String reasonText) {
+            return Mono.fromCallable {
+                responses.tryEmitComplete()
+            }.then()
         }
     }
 }

src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy

@@ -0,0 +1,90 @@
+package io.emeraldpay.dshackle.test
+
+import com.fasterxml.jackson.databind.util.ByteBufferBackedInputStream
+import org.java_websocket.WebSocket
+import org.java_websocket.handshake.ClientHandshake
+import org.java_websocket.server.WebSocketServer
+import org.joda.time.format.DateTimeFormat
+
+import java.nio.ByteBuffer
+import java.time.Instant
+import java.time.LocalDate
+import java.time.ZoneId
+import java.time.format.DateTimeFormatter
+import java.time.format.DateTimeFormatterBuilder
+
+class MockWSServer extends WebSocketServer {
+
+    private def format = DateTimeFormatter.ofPattern("HH:mm:ss.SSS")
+
+    List<ReceivedMessage> received = []
+    private WebSocket conn
+
+    private String next
+
+    MockWSServer(int port) {
+        super(new InetSocketAddress("127.0.0.1", port))
+    }
+
+    void log(String msg) {
+        println(format.format(Instant.now().atZone(ZoneId.systemDefault())) + " MOCKWS: " + msg)
+    }
+
+    void reply(String message) {
+        log(">> $message")
+        if (conn == null) {
+            log("MOCKWS: ERROR, no active connection")
+        }
+        conn.send(message)
+    }
+
+    void onNextReply(String message) {
+        next = message
+    }
+
+    @Override
+    void onOpen(WebSocket conn, ClientHandshake handshake) {
+        this.conn = conn
+        log("Opened connection from ${conn.remoteSocketAddress}")
+    }
+
+    @Override
+    void onClose(WebSocket conn, int code, String reason, boolean remote) {
+        this.conn = null
+        log("Connection closed, code ${code} with msg '${reason}' ${remote ? 'by remote' : 'by server'}")
+    }
+
+    @Override
+    void onMessage(WebSocket conn, String message) {
+        log("<< $message")
+        received.add(new ReceivedMessage(message))
+        if (next != null) {
+            reply(next)
+            next = null
+        }
+    }
+
+    @Override
+    void onMessage(WebSocket conn, ByteBuffer message) {
+        onMessage(conn, new ByteBufferBackedInputStream(message).text)
+    }
+
+    @Override
+    void onError(WebSocket conn, Exception ex) {
+        log("ERROR, $ex.message")
+        received.add(new ReceivedMessage("Err: ${ex.message}"))
+    }
+
+    @Override
+    void onStart() {
+        log("Server started")
+    }
+
+    class ReceivedMessage {
+        final String value
+
+        ReceivedMessage(String value) {
+            this.value = value
+        }
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy

@@ -0,0 +1,164 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.test.MockWSServer
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import reactor.test.StepVerifier
+import spock.lang.Shared
+import spock.lang.Specification
+
+import java.time.Duration
+
+class EthereumWsFactoryRealSpec extends Specification {
+
+    static SLEEP = 500
+
+    static int port = 19900 + new Random().nextInt(100)
+    @Shared
+    MockWSServer server
+    @Shared
+    EthereumWsFactory.EthereumWs conn
+
+    def setup() {
+        if (System.getenv("CI") == "true") {
+            println("RUN IN CI ENVIRONMENT")
+            // needs large timeouts on CI where it's much slower to run
+            SLEEP = 1500
+        }
+        port++
+        server = new MockWSServer(port)
+        server.start()
+        Thread.sleep(SLEEP)
+        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null)
+    }
+
+    def cleanup() {
+        conn.close()
+        server.stop()
+    }
+
+    def "Connects to server"() {
+        when:
+        conn.connect()
+        Thread.sleep(SLEEP)
+        println("verify....")
+        def act = server.received
+        then:
+        act.size() > 0
+        act[0].value.contains("\"method\":\"eth_subscribe\"")
+        act[0].value.contains("\"params\":[\"newHeads\"]")
+    }
+
+    def "Makes RPC request"() {
+        when:
+        conn.connect()
+        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        then:
+        StepVerifier.create(resp)
+                .then {
+                    server.onNextReply('{"jsonrpc":"2.0", "id":100, "result": "baz"}')
+                }
+                .expectNextMatches {
+                    it.hasResult() && it.resultAsProcessedString == "baz"
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(3))
+
+        when:
+        Thread.sleep(SLEEP)
+        def act = server.received
+        then:
+        act.size() == 2
+        act[1].value.contains("\"method\":\"foo_bar\"")
+    }
+
+    def "Reconnects after server disconnect"() {
+        when:
+        conn.connect()
+        conn.reconnectIntervalSeconds = 2
+        Thread.sleep(SLEEP)
+        server.stop()
+        Thread.sleep(SLEEP)
+        server = new MockWSServer(port)
+        server.start()
+        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        // reconnects in 2 seconds, give 1 extra
+        Thread.sleep(3_000)
+        def act = server.received
+
+        then:
+        act.size() > 0
+        act[0].value.contains("\"method\":\"eth_subscribe\"")
+        act[0].value.contains("\"params\":[\"newHeads\"]")
+    }
+
+    def "Gets UNAVAIL status right after disconnect"() {
+        setup:
+        def up = Mock(DefaultUpstream)
+        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null)
+        when:
+        conn.connect()
+        conn.reconnectIntervalSeconds = 10
+        Thread.sleep(SLEEP)
+        server.stop()
+        Thread.sleep(100)
+
+        then:
+        1 * up.setStatus(UpstreamAvailability.UNAVAILABLE)
+    }
+
+    def "Validates after connect"() {
+        setup:
+        def validator = Mock(EthereumUpstreamValidator)
+        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null)
+        when:
+        conn.connect()
+        Thread.sleep(100)
+
+        then:
+        1 * validator.validate()
+    }
+
+    def "Try to connects to server until it's available"() {
+        when:
+        server.stop()
+        Thread.sleep(SLEEP)
+        conn.reconnectIntervalSeconds = 1
+        conn.connect()
+        Thread.sleep(3_000)
+        server = new MockWSServer(port)
+        server.start()
+        Thread.sleep(2_000)
+        def act = server.received
+        then:
+        act.size() > 0
+        act[0].value.contains("\"method\":\"eth_subscribe\"")
+        act[0].value.contains("\"params\":[\"newHeads\"]")
+    }
+
+    def "Call after reconnect"() {
+        when:
+        conn.connect()
+        conn.reconnectIntervalSeconds = 2
+        Thread.sleep(SLEEP)
+        server.stop()
+        Thread.sleep(SLEEP)
+        server = new MockWSServer(port)
+        server.start()
+        // reconnects in 2 seconds, give 1 extra
+        Thread.sleep(3_000)
+
+        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        then:
+        StepVerifier.create(resp)
+                .then {
+                    server.onNextReply('{"jsonrpc":"2.0", "id":100, "result": "baz"}')
+                }
+                .expectNextMatches {
+                    it.hasResult() && it.resultAsProcessedString == "baz"
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(3))
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy

@@ -15,11 +15,15 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.emeraldpay.dshackle.cache.BlocksMemCache
+import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.etherjar.domain.BlockHash
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.rpc.RpcResponseError
 import io.emeraldpay.etherjar.rpc.json.BlockJson
+import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import reactor.core.publisher.Flux
 import reactor.test.StepVerifier
@@ -34,7 +38,6 @@ class EthereumWsFactorySpec extends Specification {
     def "Fetch block"() {
         setup:
         def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
-        def blocksCache = Mock(BlocksMemCache)
 
         def block = new BlockJson<TransactionRefJson>()
         block.number = 100
@@ -44,20 +47,98 @@ class EthereumWsFactorySpec extends Specification {
         block.uncles = []
         block.totalDifficulty = BigInteger.ONE
 
+        def headBlock = block.copy().tap {
+            it.transactions = null
+        }
+
         def apiMock = TestingCommons.api()
-        def upstream = TestingCommons.upstream(apiMock)
+        def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(upstream)
+        def ws = wsf.create(null, null, null)
 
         apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
 
         when:
-        def act = Flux.from(ws.getFlux())
+        Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
+        def act = Flux.from(ws.getBlocksFlux())
 
         then:
         StepVerifier.create(act)
-                .then { ws.onNewBlock(block) }
+                .then { ws.onNewHeads(headBlock).subscribe() }
                 .expectNext(BlockContainer.from(block))
                 .thenCancel()
                 .verify(Duration.ofSeconds(1))
     }
+
+    def "Makes a RPC call"() {
+        setup:
+        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def apiMock = TestingCommons.api()
+        def wsApiMock = apiMock.asWebsocket()
+        def ws = wsf.create(null, null, null)
+
+        def tx = new TransactionJson().tap {
+            hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
+        }
+        apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], tx)
+
+        when:
+        Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
+        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15))
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.id.asNumber() == 15L && Global.objectMapper.readValue(it.result, TransactionJson) == tx
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Makes a RPC call - return null"() {
+        setup:
+        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def apiMock = TestingCommons.api()
+        def wsApiMock = apiMock.asWebsocket()
+        def ws = wsf.create(null, null, null)
+
+        apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null)
+
+        when:
+        Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
+        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15))
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.id.asNumber() == 15L &&
+                            it.resultAsRawString == 'null'
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Makes a RPC call - return error"() {
+        setup:
+        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def apiMock = TestingCommons.api()
+        def wsApiMock = apiMock.asWebsocket()
+        def ws = wsf.create(null, null, null)
+
+        apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"],
+                new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test"))
+
+        when:
+        Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
+        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15))
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    it.id.asNumber() == 15L &&
+                            it.error != null &&
+                            it.error.code == RpcResponseError.CODE_METHOD_NOT_EXIST && it.error.message == "test"
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy

@@ -18,19 +18,9 @@ package io.emeraldpay.dshackle.upstream.rpcclient
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import spock.lang.Specification
 
-class JsonRpcParserSpec extends Specification {
+class ResponseRpcParserSpec extends Specification {
 
-    JsonRpcParser parser = new JsonRpcParser()
+    ResponseRpcParser parser = new ResponseRpcParser()
-
-    def "Parse just result"() {
-        setup:
-        def json = '{"result": "Hello world!"}'
-        when:
-        def act = parser.parse(json.getBytes())
-        then:
-        act.error == null
-        new String(act.result) == '"Hello world!"'
-    }
 
     def "Parse string response"() {
         setup:
@@ -178,6 +168,19 @@ class JsonRpcParserSpec extends Specification {
         !act.hasResult()
     }
 
+    def "Parse error with no result field"() {
+        setup:
+        def json = '{"jsonrpc": "2.0", "id": 1, "error": {"code": -1111, "message": "test"}}'
+        when:
+        def act = parser.parse(json.getBytes())
+        then:
+        act.error != null
+        act.error.code == -1111
+        act.error.message == "test"
+        act.hasError()
+        !act.hasResult()
+    }
+
     def "Parse error with data"() {
         setup:
         //          0       8       16                32

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy

@@ -0,0 +1,106 @@
+/**
+ * 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.rpcclient
+
+import spock.lang.Specification
+
+class ResponseWSParserSpec extends Specification {
+
+    ResponseWSParser parser = new ResponseWSParser()
+
+    def "Parse subscription response"() {
+        setup:
+        def msg = "{\n" +
+                "    \"id\": \"blocks\", \n" +
+                "    \"jsonrpc\": \"2.0\", \n" +
+                "    \"result\": \"0x9cef478923ff08bf67fde6c64013158d\"\n" +
+                "}"
+        when:
+        def act = parser.parse(msg.bytes)
+        then:
+        act.type == ResponseWSParser.Type.RPC
+        act.id.asString() == "blocks"
+        act.error == null
+        act.value == "\"0x9cef478923ff08bf67fde6c64013158d\"".bytes
+    }
+
+    def "Parse newHeads event"() {
+        setup:
+        def msg = "{\n" +
+                "  \"jsonrpc\": \"2.0\",\n" +
+                "  \"method\": \"eth_subscription\",\n" +
+                "  \"params\": {\n" +
+                "    \"result\": {\n" +
+                "      \"difficulty\": \"0x15d9223a23aa\",\n" +
+                "      \"extraData\": \"0xd983010305844765746887676f312e342e328777696e646f7773\",\n" +
+                "      \"gasLimit\": \"0x47e7c4\",\n" +
+                "      \"gasUsed\": \"0x38658\",\n" +
+                "      \"logsBloom\": \"0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000\",\n" +
+                "      \"miner\": \"0xf8b483dba2c3b7176a3da549ad41a48bb3121069\",\n" +
+                "      \"nonce\": \"0x084149998194cc5f\",\n" +
+                "      \"number\": \"0x1348c9\",\n" +
+                "      \"parentHash\": \"0x7736fab79e05dc611604d22470dadad26f56fe494421b5b333de816ce1f25701\",\n" +
+                "      \"receiptRoot\": \"0x2fab35823ad00c7bb388595cb46652fe7886e00660a01e867824d3dceb1c8d36\",\n" +
+                "      \"sha3Uncles\": \"0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347\",\n" +
+                "      \"stateRoot\": \"0xb3346685172db67de536d8765c43c31009d0eb3bd9c501c9be3229203f15f378\",\n" +
+                "      \"timestamp\": \"0x56ffeff8\",\n" +
+                "      \"transactionsRoot\": \"0x0167ffa60e3ebc0b080cdb95f7c0087dd6c0e61413140e39d94d3468d7c9689f\"\n" +
+                "    },\n" +
+                "    \"subscription\": \"0x9ce59a13059e417087c02d3236a0b1cc\"\n" +
+                "  }\n" +
+                "}"
+        when:
+        def act = parser.parse(msg.bytes)
+        then:
+        act.type == ResponseWSParser.Type.SUBSCRIPTION
+        act.id.asString() == "0x9ce59a13059e417087c02d3236a0b1cc"
+        act.error == null
+        with(new String(act.value)) {
+            it.length() > 0
+            it.startsWith("{")
+            it.endsWith("}")
+            it.contains("\"difficulty\": \"0x15d9223a23aa\"")
+        }
+    }
+
+    def "Parse RPC with error"() {
+        setup:
+        def msg = "{\"jsonrpc\":\"2.0\",\"id\":151,\"error\":{\"code\":-32602,\"message\":\"invalid blocknumber\"}}"
+        when:
+        def act = parser.parse(msg.bytes)
+        then:
+        act.type == ResponseWSParser.Type.RPC
+        act.id.asNumber() == 151L
+        act.error != null
+        act.value == null
+        with(act.error) {
+            it.code == -32602
+            it.message == "invalid blocknumber"
+        }
+    }
+
+    def "Parse RPC with null result"() {
+        setup:
+        def msg = "{\"jsonrpc\":\"2.0\",\"id\":100,\"result\":null}"
+        when:
+        def act = parser.parse(msg.bytes)
+        then:
+        act.type == ResponseWSParser.Type.RPC
+        act.id.asNumber() == 100L
+        act.error == null
+        new String(act.value) == "null"
+    }
+}

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

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

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