diff --git a/gradle.properties b/gradle.properties index 8c9494f8..0087daba 100644 --- a/gradle.properties +++ b/gradle.properties @@ -14,7 +14,7 @@ springBootVersion=2.1.4.RELEASE springVersion=5.1.4.RELEASE # Our Libs -etherjarVersion=0.6.0 +etherjarVersion=0.7.0-SNAPSHOT # Testing spockVersion=1.2-groovy-2.5 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 3ce29426..497be8e3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -1,24 +1,52 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper +import io.infinitape.etherjar.rpc.RpcCall +import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.json.ResponseJson -import io.infinitape.etherjar.rpc.transport.RpcTransport +import org.slf4j.LoggerFactory import reactor.core.publisher.Mono +import java.time.Duration class EthereumUpstream( - private val rpcTransport: RpcTransport, + private val rpcClient: RpcClient, private val objectMapper: ObjectMapper ) { + private val timeout = Duration.ofSeconds(5) + private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) + fun execute(id: Int, method: String, params: List): Mono { return Mono - .fromCompletionStage(rpcTransport.execute(method, params, Any::class.java)) + .fromCompletionStage( + rpcClient.execute(RpcCall.create(method, Any::class.java, params)) + ) + .timeout(timeout) + .doOnError { t -> + log.warn("Upstream error: ${t.message}") + } .map { val resp = ResponseJson() resp.id = id resp.result = it objectMapper.writer().writeValueAsBytes(resp) } + .onErrorMap { t -> + if (RpcException::class.java.isAssignableFrom(t.javaClass)) { + t + } else { + log.warn("Convert to RPC error. Exception: ${t.message}") + RpcException(-32020, "Error reading from upstream", null, t) + } + } + .onErrorResume(RpcException::class.java) { t -> + val resp = ResponseJson() + resp.id = id + resp.error = t.error + Mono.just(objectMapper.writer().writeValueAsBytes(resp)) + } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index 26c23e58..507480d2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.DefaultRpcClient import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport import org.springframework.beans.factory.annotation.Autowired import org.springframework.core.env.Environment @@ -33,7 +34,7 @@ class Upstreams( private fun buildClient(url: String): EthereumUpstream { return EthereumUpstream( - DefaultRpcTransport(URI(url)), + DefaultRpcClient(DefaultRpcTransport(URI(url))), objectMapper ) }