From 77685c314a9b4e438640c684083eca86248b2cae Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 11 Oct 2019 22:54:12 -0400 Subject: [PATCH] solution: use reactor-based rpc client --- .../dshackle/upstream/ConfiguredUpstreams.kt | 13 +++--- .../dshackle/upstream/UpstreamValidator.kt | 43 +++++++++++++------ .../upstream/ethereum/DirectEthereumApi.kt | 10 ++--- .../dshackle/upstream/ethereum/EthereumApi.kt | 6 +++ .../upstream/ethereum/EthereumRpcHead.kt | 25 +++++++---- .../dshackle/upstream/grpc/GrpcUpstream.kt | 22 +++++----- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 11 ++--- .../dshackle/rpc/NativeCallSpec.groovy | 14 +++--- .../dshackle/rpc/TrackAddressSpec.groovy | 6 +-- .../dshackle/rpc/TrackTxSpec.groovy | 14 +++--- .../dshackle/test/EthereumApiMock.groovy | 3 +- .../dshackle/test/EthereumApiStub.groovy | 38 ++++++---------- .../dshackle/test/TestingCommons.groovy | 3 +- .../upstream/CurrentUpstreamsSpec.groovy | 19 ++++---- .../dshackle/upstream/FilteredApisSpec.groovy | 9 ++-- .../upstream/grpc/GrpcUpstreamSpec.groovy | 16 +++---- 16 files changed, 131 insertions(+), 121 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index dbd70d8b..b7487527 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -24,8 +24,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.DefaultRpcClient -import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport +import io.infinitape.etherjar.rpc.http.ReactorHttpRpcClient import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -132,18 +131,18 @@ open class ConfiguredUpstreams( currentUpstreams.getDefaultMethods(chain) } conn.rpc?.let { endpoint -> - val rpcTransport = DefaultRpcTransport(endpoint.url) + val rpcClient = ReactorHttpRpcClient.newBuilder() + .setTarget(endpoint.url) conn.rpc?.basicAuth?.let { auth -> - rpcTransport.setBasicAuth(auth.username, auth.password) + rpcClient.setBasicAuth(auth.username, auth.password) } conn.rpc?.tls?.let { tls -> tls.ca?.let { ca -> - fileResolver.resolve(ca).inputStream().use { cert -> rpcTransport.setTrustedCertificate(cert) } + fileResolver.resolve(ca).inputStream().use { cert -> rpcClient.setTrustedCertificate(cert) } } } - val rpcClient = DefaultRpcClient(rpcTransport) rpcApi = DirectEthereumApi( - rpcClient, + rpcClient.build(), objectMapper, methods ).apply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt index 8f4a5288..4997d079 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt @@ -20,33 +20,50 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.ReactorBatch +import org.slf4j.LoggerFactory +import org.springframework.scheduling.concurrent.CustomizableThreadFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers import java.time.Duration +import java.util.concurrent.Executors import java.util.concurrent.TimeUnit class UpstreamValidator( private val ethereumUpstream: EthereumUpstream, private val options: UpstreamsConfig.Options ) { + companion object { + private val log = LoggerFactory.getLogger(UpstreamValidator::class.java) + val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("validator"))) + } fun validate(): Mono { - val batch = Batch() - val peerCount = batch.add(Commands.net().peerCount()) - val syncing = batch.add(Commands.eth().syncing()) + val batch = ReactorBatch() + val peerCount = batch.add(Commands.net().peerCount()).result + val syncing = batch.add(Commands.eth().syncing()).result return ethereumUpstream.getApi(Selector.empty) - .map { api -> api.rpcClient.execute(batch) } - .flatMap { Mono.fromCompletionStage(it) } - .timeout(Defaults.timeout) - .map { - if (syncing.get().isSyncing) { - UpstreamAvailability.SYNCING - } else if (options.minPeers != null && peerCount.get() < options.minPeers!!) { - UpstreamAvailability.IMMATURE + .subscribeOn(scheduler) + .flatMapMany { api -> api.rpcClient.execute(batch) } + .timeout(Defaults.timeout, Mono.error(Exception("Validation timeout"))) + .then(syncing) + .flatMap { value -> + if (value.isSyncing) { + Mono.just(UpstreamAvailability.SYNCING) } else { - UpstreamAvailability.OK + peerCount.map { count -> + val minPeers = options.minPeers ?: 1 + if (count < minPeers) { + UpstreamAvailability.IMMATURE + } else { + UpstreamAvailability.OK + } + } } - }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE } + } + .doOnError { err -> log.warn("Failed to validate upstream", err)} + .onErrorReturn(UpstreamAvailability.UNAVAILABLE) } fun start(): Flux { 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 a0d62dfe..08485de1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt @@ -18,8 +18,9 @@ package io.emeraldpay.dshackle.upstream.ethereum import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.upstream.CallMethods +import io.infinitape.etherjar.rpc.ReactorBatch +import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcCall -import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.ResponseJson import org.slf4j.LoggerFactory @@ -27,7 +28,7 @@ import reactor.core.publisher.Mono import java.time.Duration open class DirectEthereumApi( - val rpcClient: RpcClient, + val rpcClient: ReactorRpcClient, private val objectMapper: ObjectMapper, val targets: CallMethods ): EthereumApi(objectMapper) { @@ -68,8 +69,7 @@ open class DirectEthereumApi( } private fun callUpstream(method: String, params: List): Mono { - return Mono.fromCompletionStage( - rpcClient.execute(RpcCall.create(method, Any::class.java, params)) - ).timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout"))) + return rpcClient.execute(RpcCall.create(method, Any::class.java, params)) + .timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout"))) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt index 6093f217..fcba2a7f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.rpc.* +import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.io.InputStream @@ -25,6 +26,10 @@ abstract class EthereumApi( objectMapper: ObjectMapper ) { + companion object { + private val log = LoggerFactory.getLogger(EthereumApi::class.java) + } + private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) var upstream: Upstream? = null @@ -40,5 +45,6 @@ abstract class EthereumApi( return execute(0, rpcCall.method, rpcCall.params as List) .flatMap(convertToJS) .map(rpcCall.converter::apply) + .doOnError { err -> log.debug("Failed to read from upstream", err) } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index d8b8ac9c..7b846e35 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -18,35 +18,44 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Defaults import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.ReactorBatch import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle +import org.springframework.scheduling.concurrent.CustomizableThreadFactory import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers import java.time.Duration +import java.util.concurrent.Executors class EthereumRpcHead( private val api: DirectEthereumApi, private val interval: Duration = Duration.ofSeconds(10) ): DefaultEthereumHead(), Lifecycle { + companion object { + val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("ethereum-rpc-head"))) + } + private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java) private var refreshSubscription: Disposable? = null override fun start() { val base = Flux.interval(interval) + .publishOn(scheduler) .flatMap { - val batch = Batch() - val f = batch.add(Commands.eth().blockNumber) - api.rpcClient.execute(batch) - Mono.fromCompletionStage(f).timeout(Defaults.timeout, Mono.empty()) + api.rpcClient + .execute(Commands.eth().blockNumber) + .subscribeOn(scheduler) + .timeout(Defaults.timeout, Mono.error(Exception("Block number not received"))) } .flatMap { - val batch = Batch() - val f = batch.add(Commands.eth().getBlock(it)) - api.rpcClient.execute(batch) - Mono.fromCompletionStage(f).timeout(Defaults.timeout, Mono.empty()) + api.rpcClient + .execute(Commands.eth().getBlock(it)) + .subscribeOn(scheduler) + .timeout(Defaults.timeout, Mono.error(Exception("Block data not received"))) } .onErrorContinue { err, _ -> log.debug("RPC error ${err.message}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index 42a9ac69..430c463e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -30,7 +30,7 @@ import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.* -import io.infinitape.etherjar.rpc.emerald.EmeraldGrpcTransport +import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle @@ -38,7 +38,6 @@ import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.toMono -import java.lang.Exception import java.math.BigInteger import java.time.Duration import java.util.* @@ -50,9 +49,9 @@ import kotlin.collections.ArrayList open class GrpcUpstream( private val parentId: String, private val chain: Chain, - private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, + private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, - private val grpcTransport: EmeraldGrpcTransport + private val rpcClient: ReactorEmeraldClient ): DefaultUpstream(), Lifecycle { private var allLabels: Collection = ArrayList() @@ -68,11 +67,10 @@ open class GrpcUpstream( open fun createApi(matcher: Selector.Matcher): DirectEthereumApi { val targets = this.getMethods() - val transport = Selector.extractLabels(matcher)?.let { selector -> - grpcTransport.copyWithSelector(selector.asProto()) - } ?: grpcTransport - val rpcClient = DefaultRpcClient(transport) - return DirectEthereumApi(rpcClient, objectMapper, targets).let { + val client = Selector.extractLabels(matcher)?.let { selector -> + rpcClient.copyWithSelector(selector.asProto()) + } ?: rpcClient + return DirectEthereumApi(client, objectMapper, targets).let { it.upstream = this it } @@ -91,10 +89,10 @@ open class GrpcUpstream( val retry: Function, Flux> = Function { setStatus(UpstreamAvailability.UNAVAILABLE) - client.subscribeHead(chainRef) + blockchainStub.subscribeHead(chainRef) } - val flux = client.subscribeHead(chainRef) + val flux = blockchainStub.subscribeHead(chainRef) .compose(GrpcRetry.ManyToMany.retryAfter(retry, Duration.ofSeconds(5))) observeHead(flux) } @@ -136,7 +134,7 @@ open class GrpcUpstream( } } }.onErrorContinue { err, _ -> - log.error("Head subscription error: ${err.message}") + log.error("Head subscription error. ${err.javaClass.name}:${err.message}", err) }.doOnNext { setStatus(UpstreamAvailability.OK) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 8936c52b..bc32459b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -26,7 +26,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamChange import io.emeraldpay.grpc.Chain import io.grpc.ManagedChannelBuilder import io.grpc.netty.NettyChannelBuilder -import io.infinitape.etherjar.rpc.emerald.EmeraldGrpcTransport +import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import io.netty.handler.ssl.* import org.apache.commons.lang3.StringUtils import org.apache.commons.lang3.exception.ExceptionUtils @@ -56,7 +56,7 @@ class GrpcUpstreams( private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null private val known = HashMap() private val lock = ReentrantLock() - private var grpcTransport: EmeraldGrpcTransport? = null + private var grpcTransport: ReactorEmeraldClient? = null fun start(): Flux { val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) { @@ -73,12 +73,9 @@ class GrpcUpstreams( val client = ReactorBlockchainGrpc.newReactorStub(channel.build()) this.client = client - var i = 0 - val grpcExecutor = Executors.newCachedThreadPool { r -> Thread(r, "grpc-up-$id-${i++}") }; - this.grpcTransport = EmeraldGrpcTransport.newBuilder() + this.grpcTransport = ReactorEmeraldClient.newBuilder() .forChannel(client.channel) .setObjectMapper(objectMapper) - .setExecutorService(grpcExecutor) .build() val statusSubscription = AtomicReference() @@ -109,8 +106,6 @@ class GrpcUpstreams( prev?.dispose() subscription } - }.doFinally { - grpcExecutor.shutdown() } return updates diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 8ea5a2d9..9e793f95 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -26,7 +26,7 @@ import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.ReactorRpcClient import reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Specification @@ -42,7 +42,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new AlwaysQuorum()) def upstreams = Stub(Upstreams) - RpcClient rpcClient = Stub(RpcClient) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api(rpcClient) apiMock.upstream = Stub(Upstream) @@ -67,7 +67,7 @@ class NativeCallSpec extends Specification { def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) def upstreams = Stub(Upstreams) - RpcClient rpcClient = Stub(RpcClient) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api(rpcClient) apiMock.upstream = Stub(Upstream) @@ -95,7 +95,7 @@ class NativeCallSpec extends Specification { def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) def upstreams = Stub(Upstreams) - RpcClient rpcClient = Stub(RpcClient) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api(rpcClient) apiMock.upstream = Stub(Upstream) @@ -121,7 +121,7 @@ class NativeCallSpec extends Specification { def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) def upstreams = Stub(Upstreams) - RpcClient rpcClient = Stub(RpcClient) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api(rpcClient) apiMock.upstream = Stub(Upstream) @@ -146,7 +146,7 @@ class NativeCallSpec extends Specification { def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) def upstreams = Stub(Upstreams) - RpcClient rpcClient = Stub(RpcClient) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api(rpcClient) apiMock.upstream = Stub(Upstream) @@ -307,7 +307,7 @@ class NativeCallSpec extends Specification { def quorum = Spy(new AlwaysQuorum()) def upstreams = Stub(Upstreams) - RpcClient rpcClient = Stub(RpcClient) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api(rpcClient) apiMock.upstream = Stub(Upstream) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy index 15b2a9fb..a96ea73f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy @@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.BlockHash -import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.json.BlockJson import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor @@ -56,7 +56,7 @@ class TrackAddressSpec extends Specification { .setBalance("1234567890") .build() - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackAddress trackAddress = new TrackAddress(upstreams, Schedulers.immediate()) @@ -98,7 +98,7 @@ class TrackAddressSpec extends Specification { } def blocksBus = TopicProcessor.create() - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackAddress trackAddress = new TrackAddress(upstreams, Schedulers.immediate()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy index bd820a0e..0beddfd5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy @@ -24,7 +24,7 @@ import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionJson import reactor.core.publisher.Mono @@ -87,7 +87,7 @@ class TrackTxSpec extends Specification { .setTimestamp(blockJson.timestamp.getTime()) ).build() - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) @@ -119,7 +119,7 @@ class TrackTxSpec extends Specification { .setMined(false) .build() - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) @@ -176,7 +176,7 @@ class TrackTxSpec extends Specification { it } - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) @@ -275,7 +275,7 @@ class TrackTxSpec extends Specification { ) - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) @@ -316,7 +316,7 @@ class TrackTxSpec extends Specification { def "Tracked after first load"() { setup: - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) @@ -337,7 +337,7 @@ class TrackTxSpec extends Specification { def "Update of last notified keeps everything else"() { setup: - def apiMock = TestingCommons.api(Stub(RpcClient)) + def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index ed4688b0..54dbbe28 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver +import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.json.ResponseJson @@ -36,7 +37,7 @@ class EthereumApiMock extends DirectEthereumApi { List predefined = [] private ObjectMapper objectMapper - EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) { + EthereumApiMock(@NotNull ReactorRpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) { super(rpcClient, objectMapper, new DirectCallMethods()) this.objectMapper = objectMapper } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy index e1515c3a..977303eb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy @@ -16,14 +16,14 @@ package io.emeraldpay.dshackle.test import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.dshackle.upstream.CallMethods import io.emeraldpay.dshackle.upstream.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi -import io.infinitape.etherjar.rpc.Batch -import io.infinitape.etherjar.rpc.ExecutableBatch +import io.infinitape.etherjar.rpc.ReactorBatch +import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcCall -import io.infinitape.etherjar.rpc.RpcClient -import io.infinitape.etherjar.rpc.transport.BatchStatus +import io.infinitape.etherjar.rpc.RpcCallResponse +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import java.util.concurrent.CompletableFuture @@ -31,7 +31,7 @@ class EthereumApiStub extends DirectEthereumApi { private String id private static ObjectMapper objectMapper = TestingCommons.objectMapper() - private static RpcClient rpcClient = new RpcClientMock(); + private static ReactorRpcClient rpcClient = new RpcClientMock(); EthereumApiStub(Integer id) { this(id.toString()) @@ -47,30 +47,16 @@ class EthereumApiStub extends DirectEthereumApi { return "API Stub $id" } - static class RpcClientMock implements RpcClient { + static class RpcClientMock implements ReactorRpcClient { + @Override - CompletableFuture execute(Batch batch) { - return null + Flux execute(ReactorBatch batch) { + return Flux.error(new Exception("Not implemented in mock")) } @Override - def CompletableFuture execute(RpcCall call) { - return null - } - - @Override - ExecutableBatch batch() { - return null - } - - @Override - EthCommands eth() { - return null - } - - @Override - TraceCommands trace() { - return null + def Mono execute(RpcCall call) { + return Mono.error(new Exception("Not implemented in mock")) } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index f507f5e9..9878b3bd 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcClient import java.text.SimpleDateFormat @@ -47,7 +48,7 @@ class TestingCommons { return objectMapper } - static EthereumApiMock api(RpcClient rpcClient) { + static EthereumApiMock api(ReactorRpcClient rpcClient) { return new EthereumApiMock(rpcClient, objectMapper(), Chain.ETHEREUM) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy index dbd9e2e1..a772fab3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy @@ -2,9 +2,8 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.ReactorRpcClient import spock.lang.Specification class CurrentUpstreamsSpec extends Specification { @@ -12,7 +11,7 @@ class CurrentUpstreamsSpec extends Specification { def "add upstream"() { setup: def current = new CurrentUpstreams(TestingCommons.objectMapper()) - def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) when: current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED)) then: @@ -23,9 +22,9 @@ class CurrentUpstreamsSpec extends Specification { def "add multiple upstreams"() { setup: def current = new CurrentUpstreams(TestingCommons.objectMapper()) - def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) - def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(RpcClient))) - def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) + def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(ReactorRpcClient))) + def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) when: current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) @@ -39,10 +38,10 @@ class CurrentUpstreamsSpec extends Specification { def "remove upstream"() { setup: def current = new CurrentUpstreams(TestingCommons.objectMapper()) - def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) - def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(RpcClient))) - def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) - def up1_del = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(RpcClient))) + def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) + def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(ReactorRpcClient))) + def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) + def up1_del = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) when: current.update(new UpstreamChange(Chain.ETHEREUM, up1, UpstreamChange.ChangeType.ADDED)) current.update(new UpstreamChange(Chain.ETHEREUM_CLASSIC, up2, UpstreamChange.ChangeType.ADDED)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index d0ffbc58..985a4230 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -22,8 +22,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.DefaultRpcClient -import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.ReactorRpcClient import reactor.test.StepVerifier import spock.lang.Retry import spock.lang.Specification @@ -32,7 +31,7 @@ import java.time.Duration class FilteredApisSpec extends Specification { - def rpcClient = new DefaultRpcClient(null) + def rpcClient = Stub(ReactorRpcClient) def objectMapper = TestingCommons.objectMapper() def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) @@ -141,8 +140,8 @@ class FilteredApisSpec extends Specification { def "Makes pause between batches"() { when: - def api1 = TestingCommons.api(Stub(RpcClient)) - def api2 = TestingCommons.api(Stub(RpcClient)) + def api1 = TestingCommons.api(Stub(ReactorRpcClient)) + def api2 = TestingCommons.api(Stub(ReactorRpcClient)) def up1 = TestingCommons.upstream(api1) def up2 = TestingCommons.upstream(api2) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy index c7f501ca..f0c3fef1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy @@ -29,8 +29,8 @@ import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.JacksonRpcConverter -import io.infinitape.etherjar.rpc.RpcClient -import io.infinitape.etherjar.rpc.emerald.EmeraldGrpcTransport +import io.infinitape.etherjar.rpc.ReactorRpcClient +import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import io.infinitape.etherjar.rpc.json.BlockJson import reactor.test.StepVerifier import spock.lang.Specification @@ -48,7 +48,7 @@ class GrpcUpstreamSpec extends Specification { setup: def callData = [:] def chain = Chain.ETHEREUM - def api = TestingCommons.api(Stub(RpcClient)) + def api = TestingCommons.api(Stub(ReactorRpcClient)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -74,7 +74,7 @@ class GrpcUpstreamSpec extends Specification { ) } }) - def transport = EmeraldGrpcTransport.newBuilder().forChannel(client.channel).build() + def transport = ReactorEmeraldClient.newBuilder().forChannel(client.channel).build() def upstream = new GrpcUpstream("test", chain, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() @@ -91,7 +91,7 @@ class GrpcUpstreamSpec extends Specification { def "Follows difficulty, ignores less difficult"() { setup: - def api = TestingCommons.api(Stub(RpcClient)) + def api = TestingCommons.api(Stub(ReactorRpcClient)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -130,7 +130,7 @@ class GrpcUpstreamSpec extends Specification { ) } }) - def transport = EmeraldGrpcTransport.newBuilder().forChannel(client.channel).build() + def transport = ReactorEmeraldClient.newBuilder().forChannel(client.channel).build() def upstream = new GrpcUpstream("test", Chain.ETHEREUM, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() @@ -150,7 +150,7 @@ class GrpcUpstreamSpec extends Specification { def callData = [:] def finished = new CompletableFuture() def chain = Chain.ETHEREUM - def api = TestingCommons.api(Stub(RpcClient)) + def api = TestingCommons.api(Stub(ReactorRpcClient)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -190,7 +190,7 @@ class GrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def transport = EmeraldGrpcTransport.newBuilder().forChannel(client.channel).build() + def transport = ReactorEmeraldClient.newBuilder().forChannel(client.channel).build() def upstream = new GrpcUpstream("test", chain, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()