From 2a93b9efa3336db55d727cac840ba939d2c47f4d Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 5 Nov 2019 22:11:47 -0500 Subject: [PATCH] problem: stuck if upstream returned null --- .../upstream/ethereum/DirectEthereumApi.kt | 18 +++- .../dshackle/test/TestingCommons.groovy | 8 ++ .../ethereum/DirectEthereumApiSpec.groovy | 92 +++++++++++++++++++ 3 files changed, 117 insertions(+), 1 deletion(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt index 52faa582..757163bf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt @@ -24,6 +24,7 @@ import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.ResponseJson import org.slf4j.LoggerFactory import reactor.core.publisher.Mono +import reactor.core.publisher.switchIfEmpty import java.time.Duration open class DirectEthereumApi( @@ -41,6 +42,10 @@ open class DirectEthereumApi( targets.isAllowed(method) -> callUpstream(method, params) else -> Mono.error(RpcException(-32601, "Method not allowed or not found")) } + return processResult(id, method, result) + } + + public fun processResult(id: Int, method: String, result: Mono): Mono { return result .doOnError { t -> log.warn("Upstream error: [${t.message}] for $method") @@ -49,7 +54,18 @@ open class DirectEthereumApi( val resp = ResponseJson() resp.id = id resp.result = it - objectMapper.writer().writeValueAsBytes(resp) + resp + } + .switchIfEmpty( + Mono.fromCallable { + val resp = ResponseJson() + resp.id = id + resp.result = null + resp + } + ) + .map { + objectMapper.writer().writeValueAsBytes(it) } .onErrorResume(StatusRuntimeException::class.java) { t -> if (t.status.code == Status.Code.CANCELLED) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 9878b3bd..4d8ce975 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -60,6 +60,14 @@ class TestingCommons { return new EthereumUpstreamMock(Chain.ETHEREUM, api) } + static EthereumUpstreamMock upstream(DirectEthereumApi api, String method) { + return upstream(api, [method]) + } + + static EthereumUpstreamMock upstream(DirectEthereumApi api, List methods) { + return new EthereumUpstreamMock(Chain.ETHEREUM, api, new DirectCallMethods(methods)) + } + static AggregatedUpstream aggregatedUpstream(DirectEthereumApi api) { return aggregatedUpstream(upstream(api)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy new file mode 100644 index 00000000..6374b569 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy @@ -0,0 +1,92 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.DirectCallMethods +import io.infinitape.etherjar.rpc.ReactorRpcClient +import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError +import reactor.core.publisher.Mono +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration + +class DirectEthereumApiSpec extends Specification { + + DirectEthereumApi api = new DirectEthereumApi(Stub(ReactorRpcClient), TestingCommons.objectMapper(), new DirectCallMethods()) + + def "Process successful result"() { + setup: + def result = Mono.just("hello") + when: + def act = api.processResult(1, "eth_test", result) + .map { new String(it) } + + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":1,"result":"hello"}') + .expectComplete() + .verify(Duration.ofSeconds(1)) + + } + + def "Process empty result"() { + setup: + def result = Mono.empty() + when: + def act = api.processResult(1, "eth_test", result) + .map { new String(it) } + + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":1,"result":null}') + .expectComplete() + .verify(Duration.ofSeconds(1)) + + } + + def "Process standard RPC error"() { + setup: + def result = Mono.error(new RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Test Error", Map.of("foo", "bar"))) + when: + def act = api.processResult(1, "eth_test", result) + .map { new String(it) } + + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"Test Error","data":{"foo":"bar"}}}') + .expectComplete() + .verify(Duration.ofSeconds(1)) + + } + + def "Process internal exception"() { + setup: + def result = Mono.error(new InterruptedException("test")) + when: + def act = api.processResult(1, "eth_test", result) + .map { new String(it) } + + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":1,"error":{"code":-32020,"message":"Error reading from upstream"}}') + .expectComplete() + .verify(Duration.ofSeconds(1)) + + } +}