solution: Ethereum WS connection fetches block through the same WS channel, instead of HTTP RPC

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/upstream/ethereum/EthereumRpcUpstream.kt

@@ -79,7 +79,7 @@ open class EthereumRpcUpstream(
 
     open fun createHead(): Head {
         return if (ethereumWsFactory != null) {
-            val ws = ethereumWsFactory.create(this).apply {
+            val ws = ethereumWsFactory.create().apply {
                 connect()
             }
             val wsHead = EthereumWsHead(ws).apply {

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

@@ -23,25 +23,33 @@ import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
 import io.emeraldpay.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 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.AtomicInteger
+
 
 class EthereumWsFactory(
         private val uri: URI,
@@ -50,27 +58,39 @@ class EthereumWsFactory(
 
     var basicAuth: AuthConfig.ClientBasicAuth? = null
 
-    fun create(upstream: EthereumUpstream): EthereumWs {
+    fun create(): EthereumWs {
-        return EthereumWs(uri, origin, upstream, basicAuth)
+        return EthereumWs(uri, origin, basicAuth)
     }
 
     class EthereumWs(
             private val uri: URI,
             private val origin: URI,
-            private val upstream: EthereumUpstream,
             private val basicAuth: AuthConfig.ClientBasicAuth?
     ) : 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\"]}"
         }
 
-        private val topic = Sinks
+        private val parser = ResponseWSParser()
+
+        private val blocks = Sinks
                 .many()
                 .multicast()
                 .directBestEffort<BlockContainer>()
+        private val 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
 
@@ -89,11 +109,6 @@ class EthereumWsFactory(
         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()
                     .doOnError(
                             { _, t ->
@@ -101,9 +116,7 @@ class EthereumWsFactory(
                                 // going to try to reconnect later
                                 tryReconnectLater()
                             },
-                            { _, _ ->
+                            { _, _ -> }
-
-                            }
                     )
                     .headers { headers ->
                         headers.add(HttpHeaderNames.ORIGIN, origin)
@@ -114,11 +127,7 @@ class EthereumWsFactory(
                         }
                     }
                     .let {
-                        if (uri.scheme == "wss") {
+                        if (uri.scheme == "wss") it.secure() else it
-                            it.secure()
-                        } else {
-                            it
-                        }
                     }
                     .websocket(
                             WebsocketClientSpec.builder()
@@ -128,29 +137,31 @@ class EthereumWsFactory(
                     )
                     .uri(uri)
                     .handle { inbound, outbound ->
+                        handle(inbound, outbound)
+                    }
+                    .doOnError {
+                        log.error("Failed to setup WS connection", it)
+                    }
+                    .subscribe()
+        }
+
+        fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
             val consumer = inbound.aggregateFrames()
-                                .aggregateFrames(8 * 65_536)
+                    // accept up to 1Mb messages
+                    .aggregateFrames(16 * 65_536)
                     .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 ->
@@ -161,24 +172,58 @@ class EthereumWsFactory(
                         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 +233,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 +242,53 @@ 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 {
+                // 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)
+            }.flatMap { request ->
+                waitForResponse(request.t1, request.t2)
+            }
+        }
+
+        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): 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()
+                                .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) }
+                    }
+        }
+
+        fun getBlocksFlux(): Flux<BlockContainer> {
+            return this.blocks.asFlux()
         }
 
         override fun close() {
@@ -216,5 +299,4 @@ class EthereumWsFactory(
     }
 
 
-
 }

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

@@ -34,7 +34,7 @@ class EthereumWsHead(
 
     override fun start() {
         this.subscription?.dispose()
-        this.subscription = super.follow(ws.getFlux())
+        this.subscription = super.follow(ws.getBlocksFlux())
     }
 
     override fun stop() {

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/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/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()
 
         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()
+
+        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()
+
+        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()
+
+        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"
+    }
+}