From 568a045271730ac2acc769f2ec7a84d620fba990 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 19 Sep 2021 22:52:21 -0400 Subject: [PATCH] solution: Ethereum WS connection fetches block through the same WS channel, instead of HTTP RPC --- .../kotlin/io/emeraldpay/dshackle/Global.kt | 2 + .../upstream/ethereum/EthereumRpcUpstream.kt | 2 +- .../upstream/ethereum/EthereumWsFactory.kt | 222 +++++++++++------ .../upstream/ethereum/EthereumWsHead.kt | 2 +- .../upstream/ethereum/NativeCallRouter.kt | 6 +- .../upstream/rpcclient/JsonRpcError.kt | 4 + .../upstream/rpcclient/JsonRpcHttpClient.kt | 4 +- .../upstream/rpcclient/JsonRpcParser.kt | 111 --------- .../upstream/rpcclient/JsonRpcRequest.kt | 53 ++-- .../upstream/rpcclient/JsonRpcResponse.kt | 11 + .../upstream/rpcclient/JsonRpcWsClient.kt | 30 +++ .../upstream/rpcclient/ResponseParser.kt | 182 ++++++++++++++ .../upstream/rpcclient/ResponseRpcParser.kt | 36 +++ .../upstream/rpcclient/ResponseWSParser.kt | 110 ++++++++ .../dshackle/test/EthereumApiMock.groovy | 234 +++++++++++++++++- .../ethereum/EthereumWsFactorySpec.groovy | 93 ++++++- ...ec.groovy => ResponseRpcParserSpec.groovy} | 27 +- .../rpcclient/ResponseWSParserSpec.groovy | 106 ++++++++ 18 files changed, 1003 insertions(+), 232 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt rename src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/{JsonRpcParserSpec.groovy => ResponseRpcParserSpec.groovy} (94%) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index dfdc9aa6..db3c8da7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 1e0a9600..1844fa49 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index e8e44c27..37977e31 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/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 { - return EthereumWs(uri, origin, upstream, basicAuth) + fun create(): EthereumWs { + 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() + private val rpcSend = Sinks + .many() + .unicast() + .onBackpressureBuffer() + private val rpcReceive = Sinks + .many() + .multicast() + .directBestEffort() + 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("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") { - it.secure() - } else { - it - } + if (uri.scheme == "wss") it.secure() else it } .websocket( WebsocketClientSpec.builder() @@ -128,57 +137,93 @@ class EthereumWsFactory( ) .uri(uri) .handle { inbound, outbound -> - val consumer = inbound.aggregateFrames() - .aggregateFrames(8 * 65_536) - .receiveFrames() - .flatMap { - val msg: SubscriptionJson = objectMapper.readerFor(SubscriptionJson::class.java) - .readValue(ByteBufInputStream(it.content()) as InputStream) - when { - msg.error != null -> { - Mono.error(IllegalStateException("Received error from WS upstream")) - } - msg.subscription == subscriptionId.get() -> { - onNewBlock(msg.blockResult) - Mono.empty() - } - msg.subscription == null -> { - // received ID for subscription - subscriptionId.set(msg.result.asText()) - log.debug("Connected to $uri") - Mono.empty() - } - else -> { - Mono.error(IllegalStateException("Unknown message received: ${msg.subscription}")) - } - } - } - .onErrorResume { t -> - log.warn("Connection dropped to $uri. Error: ${t.message}") - // going to try to reconnect later - tryReconnectLater() - // completes current outbound flow - Mono.empty() - } - - - outbound.sendString(Mono.just(START_REQUEST) - .doOnError { log.warn("Failed to start WS subscription. ${it.javaClass}: ${it.message}") }) - .then(consumer.then()) + handle(inbound, outbound) } .doOnError { - println(it) + log.error("Failed to setup WS connection", it) } .subscribe() } - fun onNewBlock(block: BlockJson) { - // WS returns incomplete blocks, i.e. without some fields, so need to fetch full block data - if (block.difficulty == null || block.transactions == null) { + fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { + val consumer = inbound.aggregateFrames() + // accept up to 1Mb messages + .aggregateFrames(16 * 65_536) + .receiveFrames() + .map { ByteBufInputStream(it.content()).readAllBytes() } + .flatMap { + try { + val msg = parser.parse(it) + if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) { + onSubscription(msg) + } else { + onRpc(msg) + } + } 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}") + // going to try to reconnect later + tryReconnectLater() + // completes current outbound flow + Mono.empty() + } + + val start = Mono.just(START_REQUEST).map { + Unpooled.wrappedBuffer(it.toByteArray()) + } + val calls = rpcSend + .asFlux() + .map { + Unpooled.wrappedBuffer(Global.objectMapper.writeValueAsBytes(it)) + } + + return outbound.send( + Flux.merge( + start, + calls.subscribeOn(Schedulers.boundedElastic()), + consumer.then(Mono.empty()).subscribeOn(Schedulers.boundedElastic()) + ) + ) + } + + fun onRpc(msg: ResponseWSParser.WsResponse): Mono { + 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() + } + } + + fun onSubscription(msg: ResponseWSParser.WsResponse): Mono { + 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 + }.flatMap { onNewHeads(it) }.then() + } + + fun onNewHeads(block: BlockJson): Mono { + // 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() - .read(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false))) + call(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(5) .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500)) @@ -195,17 +242,53 @@ class EthereumWsFactory( } .timeout(Defaults.timeout, Mono.empty()) .onErrorResume { Mono.empty() } - .subscribe { - topic.tryEmitNext(it) + .doOnNext { + blocks.tryEmitNext(it) } - + .then() } else { - topic.tryEmitNext(BlockContainer.from(block)) + Mono.fromCallable { + blocks.tryEmitNext(BlockContainer.from(block)) + }.then() } } - fun getFlux(): Flux { - return this.topic.asFlux() + fun call(originalRequest: JsonRpcRequest): Mono { + 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 { + 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 { + return this.blocks.asFlux() } override fun close() { @@ -216,5 +299,4 @@ class EthereumWsFactory( } - } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 4abe9284..3386611e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/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() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt index cea25456..e05813ec 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt @@ -123,8 +123,8 @@ class NativeCallRouter( } } - fun getBlockByNumber(params: List): Mono? { - if (params.size != 2) { + fun getBlockByNumber(params: List): Mono? { + 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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt index 66791a88..50baa980 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt +++ b/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)" + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index ce50a5a6..5fbf6b04 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt deleted file mode 100644 index 17fbad91..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt +++ /dev/null @@ -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) - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt index 5339468e..bef4df17 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt +++ b/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 + val params: List, + val id: Int ) { + constructor(method: String, params: List) : 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() { + + 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) + } + + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt index 03ed29e8..63fb1e67 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt +++ b/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() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt new file mode 100644 index 00000000..9bed273d --- /dev/null +++ b/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 { + + override fun read(key: JsonRpcRequest): Mono { + return ws.call(key) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt new file mode 100644 index 00000000..af75484b --- /dev/null +++ b/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 { + + 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 + + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt new file mode 100644 index 00000000..7db80ace --- /dev/null +++ b/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() { + + 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)) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt new file mode 100644 index 00000000..07e7e654 --- /dev/null +++ b/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() { + + 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? + ) + +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index b92e3ec7..eb4224ee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/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 { @@ -57,7 +78,7 @@ class EthereumApiMock implements Reader { } @Override - Mono read(JsonRpcRequest request) { + Mono read(JsonRpcRequest request, boolean required = true) { Callable call = { def predefined = predefined.find { it.isSame(request.method, request.params) } byte[] result = null @@ -65,7 +86,7 @@ class EthereumApiMock implements Reader { 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 { 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 return Mono.fromCallable(call) } @@ -104,6 +128,10 @@ class EthereumApiMock implements Reader { responseObserver.onCompleted() } + WebsocketApi asWebsocket() { + return new WebsocketApi(this) + } + class PredefinedResponse { String method List params @@ -132,8 +160,202 @@ class EthereumApiMock implements Reader { } } - void print() { - println "Execute API: $method ${params ? params : '_'} >> $result" + void print(int id) { + println "Execute API: $id $method ${params ? params : '_'} >> $result" + } + } + + class WebsocketApi { + private final EthereumApiMock api + + private Sinks.Many responses = Sinks + .many() + .unicast() + .onBackpressureBuffer() + private Sinks.Many 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 responses + private final Flux jsonResponses + + WebsocketInboundMock(Flux responses, Flux jsonResponses) { + this.responses = responses + this.jsonResponses = jsonResponses + } + + @Override + String selectedSubprotocol() { + throw new UnsupportedOperationException() + } + + @Override + HttpHeaders headers() { + throw new UnsupportedOperationException() + } + + @Override + Mono receiveCloseStatus() { + return Mono.empty() + } + + @Override + ByteBufFlux receive() { + throw new UnsupportedOperationException() + } + + @Override + Flux receiveObject() { + throw new UnsupportedOperationException() + } + + @Override + NettyInbound withConnection(Consumer withConnection) { + return this + } + + @Override + Flux 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 responses + + WebsocketOutboundMock(EthereumApiMock api, Sinks.Many responses) { + this.api = api + this.responses = responses + } + + @Override + String selectedSubprotocol() { + throw new UnsupportedOperationException() + } + + @Override + ByteBufAllocator alloc() { + throw new UnsupportedOperationException() + } + + private void handle(Publisher 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 dataStream) { + handle(dataStream) + return this + } + + @Override + NettyOutbound send(Publisher dataStream, Predicate predicate) { + handle(dataStream) + return this + } + + @Override + NettyOutbound sendObject(Publisher dataStream, Predicate 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 NettyOutbound sendUsing(Callable sourceInput, BiFunction mappedInput, Consumer sourceCleanup) { + return this + } + + @Override + NettyOutbound withConnection(Consumer withConnection) { + return this + } + + @Override + Mono sendClose() { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() + } + + @Override + Mono sendClose(int rsv) { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() + } + + @Override + Mono sendClose(int statusCode, @Nullable String reasonText) { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() + } + + @Override + Mono sendClose(int rsv, int statusCode, @Nullable String reasonText) { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy index 32e6615d..f5415999 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy +++ b/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() 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 ws = wsf.create(upstream) + def wsApiMock = apiMock.asWebsocket() + 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)) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy similarity index 94% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy index c07f8444..f2f3542d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy +++ b/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() - - 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!"' - } + ResponseRpcParser parser = new ResponseRpcParser() 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 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy new file mode 100644 index 00000000..460e5e02 --- /dev/null +++ b/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" + } +}