diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt index ffb04461..c811b49e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt @@ -64,7 +64,8 @@ open class ReadRpcJson( } RequestJson( json["method"].toString(), - json["params"] as List<*>, + //params MAY be omitted + (json["params"] ?: emptyList()) as List<*>, id ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index c7e9504e..ba2ca4b1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -19,15 +19,12 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory -import io.emeraldpay.dshackle.quorum.QuorumRpcReader -import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain @@ -38,7 +35,6 @@ import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.core.publisher.* -import reactor.util.function.Tuples import java.lang.Exception @Service @@ -53,7 +49,7 @@ open class NativeCall( open fun nativeCall(requestMono: Mono): Flux { return requestMono.flatMapMany(this::prepareCall) - .map(this::setupCallParams) + .map(this::parseParams) .parallel() .flatMap(this::fetch) .sequential() @@ -62,7 +58,7 @@ open class NativeCall( .onErrorResume(this::processException) } - fun setupCallParams(it: CallContext): CallContext { + fun parseParams(it: CallContext): CallContext { val params = extractParams(it.payload.params) return it.withPayload(ParsedCallDetails(it.payload.method, params)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy index 02a6771f..2120c710 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy @@ -77,6 +77,101 @@ class ReadRpcJsonSpec extends Specification { reader.getType("".bytes) then: thrown(RpcException) + } + def "Parse basic"() { + when: + def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":1, "params":[]}'.bytes) + then: + act.type == ProxyCall.RpcType.SINGLE + act.ids.size() == 1 + act.ids[0] == 1 + act.items.size() == 1 + with(act.items[0]) { + id == 0 + method == "net_peerCount" + payload.toStringUtf8() == "[]" + } + } + + def "Parse basic with string id"() { + when: + def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":"ggk19K5a", "params":[]}'.bytes) + then: + act.type == ProxyCall.RpcType.SINGLE + act.ids.size() == 1 + act.ids[0] == "ggk19K5a" + act.items.size() == 1 + with(act.items[0]) { + id == 0 + method == "net_peerCount" + payload.toStringUtf8() == "[]" + } + } + + def "Parse basic without params"() { + when: + def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":2}'.bytes) + then: + act.type == ProxyCall.RpcType.SINGLE + act.ids.size() == 1 + act.ids[0] == 2 + act.items.size() == 1 + with(act.items[0]) { + id == 0 + method == "net_peerCount" + payload.toStringUtf8() == "[]" + } + } + + def "Parse with parameters"() { + when: + def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":1, "params":["0x015f"]}'.bytes) + then: + act.type == ProxyCall.RpcType.SINGLE + act.ids.size() == 1 + act.ids[0] == 1 + act.items.size() == 1 + with(act.items[0]) { + id == 0 + method == "net_peerCount" + payload.toStringUtf8() == '["0x015f"]' + } + } + + def "Parse single batch"() { + when: + def act = reader.apply('[{"jsonrpc":"2.0", "method":"net_peerCount", "id":1, "params":[]}]'.bytes) + then: + act.type == ProxyCall.RpcType.BATCH + act.ids.size() == 1 + act.ids[0] == 1 + act.items.size() == 1 + with(act.items[0]) { + id == 0 + method == "net_peerCount" + payload.toStringUtf8() == "[]" + } + } + + def "Parse multi batch"() { + when: + def act = reader.apply('[{"jsonrpc":"2.0", "method":"net_peerCount", "id":"xdd", "params":[]}, {"jsonrpc":"2.0", "method":"foo_bar", "id":4, "params":[143, false]}]'.bytes) + then: + act.type == ProxyCall.RpcType.BATCH + act.ids.size() == 2 + act.ids[0] == "xdd" + act.ids[1] == 4 + act.items.size() == 2 + with(act.items[0]) { + id == 0 + method == "net_peerCount" + payload.toStringUtf8() == "[]" + } + with(act.items[1]) { + id == 1 + method == "foo_bar" + payload.toStringUtf8() == '[143,false]' + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 4d0e4b13..49b1dcbc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -17,23 +17,21 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper +import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.quorum.BroadcastQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum -import io.emeraldpay.dshackle.quorum.NonEmptyQuorum import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.MultistreamHolder -import io.emeraldpay.dshackle.upstream.ethereum.NativeCallRouter import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcResponseError import reactor.core.publisher.Mono @@ -42,7 +40,6 @@ import spock.lang.Ignore import spock.lang.Specification import java.time.Duration -import java.util.concurrent.TimeoutException class NativeCallSpec extends Specification { @@ -107,7 +104,7 @@ class NativeCallSpec extends Specification { 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, 1)) } } - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, quorum, + def call = new NativeCall.CallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_test", [])) when: @@ -127,7 +124,7 @@ class NativeCallSpec extends Specification { 1 * read(_) >> Mono.empty() } } - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, quorum, + def call = new NativeCall.CallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_test", [])) when: @@ -179,7 +176,7 @@ class NativeCallSpec extends Specification { when: def resp = nativeCall.buildResponse( - new NativeCall.CallContext(1561, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json)) + new NativeCall.CallContext(1561, TestingCommons.multistream(TestingCommons.api()), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json)) ) then: resp.id == 1561 @@ -229,11 +226,114 @@ class NativeCallSpec extends Specification { def resp = nativeCall.prepareCall(req) then: StepVerifier.create(resp) - .expectErrorMatches({t -> t instanceof NativeCall.CallFailure && t.id == 0}) + .expectErrorMatches({ t -> t instanceof NativeCall.CallFailure && t.id == 0 }) // .expectComplete() .verify(Duration.ofSeconds(1)) } + def "Prepare call"() { + setup: + def upstreams = Mock(MultistreamHolder) + def nativeCall = new NativeCall(upstreams) + + def req = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.CHAIN_ETHEREUM) + .addItems( + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(1) + .setMethod("eth_test") + .setPayload(ByteString.copyFromUtf8("[]")) + ) + .build() + when: + def act = nativeCall.prepareCall(req, TestingCommons.emptyMultistream()) + .collectList().block(Duration.ofSeconds(1)) + then: + act.size() == 1 + with(act[0]) { + id == 1 + payload.method == "eth_test" + payload.params == "[]" + } + } + + def "Prepare call without payload"() { + setup: + def upstreams = Mock(MultistreamHolder) + def nativeCall = new NativeCall(upstreams) + + def req = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.CHAIN_ETHEREUM) + .addItems( + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(1) + .setMethod("eth_test") + ) + .build() + when: + def act = nativeCall.prepareCall(req, TestingCommons.emptyMultistream()) + .collectList().block(Duration.ofSeconds(1)) + then: + act.size() == 1 + with(act[0]) { + id == 1 + payload.method == "eth_test" + payload.params == "" + } + } + + def "Parse empty params"() { + setup: + def nativeCall = new NativeCall(Stub(MultistreamHolder)) + def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(), + new NativeCall.RawCallDetails("eth_test", "[]")) + when: + def act = nativeCall.parseParams(ctx) + then: + act.id == 1 + act.payload.params == [] + act.payload.method == "eth_test" + } + + def "Parse none params"() { + setup: + def nativeCall = new NativeCall(Stub(MultistreamHolder)) + def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(), + new NativeCall.RawCallDetails("eth_test", "")) + when: + def act = nativeCall.parseParams(ctx) + then: + act.id == 1 + act.payload.params == [] + act.payload.method == "eth_test" + } + + def "Parse single param"() { + setup: + def nativeCall = new NativeCall(Stub(MultistreamHolder)) + def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(), + new NativeCall.RawCallDetails("eth_test", "[false]")) + when: + def act = nativeCall.parseParams(ctx) + then: + act.id == 1 + act.payload.params == [false] + act.payload.method == "eth_test" + } + + def "Parse multi param"() { + setup: + def nativeCall = new NativeCall(Stub(MultistreamHolder)) + def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(), + new NativeCall.RawCallDetails("eth_test", "[false, 123]")) + when: + def act = nativeCall.parseParams(ctx) + then: + act.id == 1 + act.payload.params == [false, 123] + act.payload.method == "eth_test" + } + @Ignore //TODO def "Calls cache before remote"() { @@ -241,7 +341,7 @@ class NativeCallSpec extends Specification { def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams) def api = TestingCommons.api() - def upstream = TestingCommons.aggregatedUpstream(api) + def upstream = TestingCommons.multistream(api) def ctx = new NativeCall.CallContext(10, upstream, @@ -259,7 +359,7 @@ class NativeCallSpec extends Specification { setup: def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams) - def upstream = TestingCommons.aggregatedUpstream(TestingCommons.api()) + def upstream = TestingCommons.multistream(TestingCommons.api()) def ctx = new NativeCall.CallContext(10, upstream, diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index dade8700..9b6518ce 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -16,15 +16,12 @@ */ package io.emeraldpay.dshackle.test -import com.fasterxml.jackson.core.Version -import com.fasterxml.jackson.databind.DeserializationFeature -import com.fasterxml.jackson.databind.ObjectMapper -import com.fasterxml.jackson.databind.module.SimpleModule + import io.emeraldpay.dshackle.FileResolver -import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.CacheConfig +import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods @@ -33,9 +30,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.JacksonRpcConverter - -import java.text.SimpleDateFormat class TestingCommons { @@ -54,16 +48,20 @@ class TestingCommons { return new EthereumUpstreamMock(Chain.ETHEREUM, api, new DirectCallMethods(methods)) } - static Multistream aggregatedUpstream(Reader api) { - return aggregatedUpstream(upstream(api)) + static Multistream multistream(Reader api) { + return multistream(upstream(api)) } - static Multistream aggregatedUpstream(EthereumUpstream up) { + static Multistream multistream(EthereumUpstream up) { return new EthereumMultistream(Chain.ETHEREUM, [up], Caches.default()).tap { start() } } + static Multistream emptyMultistream() { + return multistream(new EmptyReader()) + } + static CachesFactory emptyCaches() { return new CachesFactory(new CacheConfig()) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy index fa506e91..c00c72f4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy @@ -88,7 +88,7 @@ class EthereumReaderSpec extends Specification { def api = TestingCommons.api() api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) - def upstream = TestingCommons.aggregatedUpstream(api) + def upstream = TestingCommons.multistream(api) def reader = new EthereumReader(upstream, caches, calls) when: @@ -109,7 +109,7 @@ class EthereumReaderSpec extends Specification { def api = TestingCommons.api() api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) - def upstream = TestingCommons.aggregatedUpstream(api) + def upstream = TestingCommons.multistream(api) def reader = new EthereumReader(upstream, caches, calls) when: @@ -146,7 +146,7 @@ class EthereumReaderSpec extends Specification { .build() def api = TestingCommons.api() api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) - def upstream = TestingCommons.aggregatedUpstream(api) + def upstream = TestingCommons.multistream(api) def reader = new EthereumReader(upstream, caches, calls) when: @@ -184,7 +184,7 @@ class EthereumReaderSpec extends Specification { def api = TestingCommons.api() api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson) - def upstream = TestingCommons.aggregatedUpstream(api) + def upstream = TestingCommons.multistream(api) def reader = new EthereumReader(upstream, caches, calls) when: @@ -200,7 +200,7 @@ class EthereumReaderSpec extends Specification { api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10") api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff") EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api) - def upstreams = TestingCommons.aggregatedUpstream(upstream) + def upstreams = TestingCommons.multistream(upstream) def reader = new EthereumReader(upstreams, Caches.default(), calls) reader.start() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy index 02a400e2..4e3bac26 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy @@ -17,7 +17,7 @@ class NativeCallRouterSpec extends Specification { def methods = new DefaultEthereumMethods(Chain.ETHEREUM) def router = new NativeCallRouter( new EthereumReader( - TestingCommons.aggregatedUpstream(TestingCommons.api()), + TestingCommons.multistream(TestingCommons.api()), Caches.default(), ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)) ),