From 8a99212fc56348727009218fedbdb6f00a2fc79b Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 20 Aug 2019 23:14:49 -0400 Subject: [PATCH] solution: cache latest blocks in memory --- .../dshackle/cache/BlocksMemCache.kt | 30 +++++++++ .../dshackle/reader/BlockApiReader.kt | 27 ++++++++ .../dshackle/reader/BlockCacheReader.kt | 16 +++++ .../dshackle/reader/CompoundReader.kt | 23 +++++++ .../emeraldpay/dshackle/reader/EmptyReader.kt | 10 +++ .../io/emeraldpay/dshackle/reader/Reader.kt | 9 +++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 52 +++++++++----- .../dshackle/upstream/AggregatedUpstream.kt | 46 ++++++++++++- .../dshackle/upstream/CachingEthereumApi.kt | 48 +++++++++++++ .../dshackle/upstream/ChainUpstreams.kt | 20 +++--- .../dshackle/upstream/ConfiguredUpstreams.kt | 7 +- .../dshackle/upstream/DirectEthereumApi.kt | 58 ++++++++++++++++ .../dshackle/upstream/EmptyEthereumHead.kt | 12 ++++ .../dshackle/upstream/EthereumApi.kt | 67 ++----------------- .../dshackle/upstream/EthereumRpcHead.kt | 2 +- .../dshackle/upstream/EthereumUpstream.kt | 8 +-- .../dshackle/upstream/FilteringApiIterator.kt | 4 +- .../dshackle/upstream/GrpcUpstream.kt | 6 +- .../emeraldpay/dshackle/upstream/Upstream.kt | 3 +- .../dshackle/cache/BlocksMemCacheSpec.groovy | 52 ++++++++++++++ .../dshackle/rpc/NativeCallSpec.groovy | 58 ++++++++++++++-- .../dshackle/rpc/StreamHeadSpec.groovy | 3 +- .../dshackle/test/EthereumApiMock.groovy | 5 +- .../dshackle/test/EthereumUpstreamMock.groovy | 3 +- .../dshackle/test/TestingCommons.groovy | 10 ++- .../dshackle/test/UpstreamsMock.groovy | 2 +- .../upstream/EthereumGrpcTransportSpec.groovy | 17 ++--- .../upstream/FilteringApiIteratorSpec.groovy | 2 +- 28 files changed, 476 insertions(+), 124 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/EmptyReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/Reader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectEthereumApi.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyEthereumHead.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt new file mode 100644 index 00000000..9e51d8c9 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt @@ -0,0 +1,30 @@ +package io.emeraldpay.dshackle.cache + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Mono +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ConcurrentLinkedQueue + +class BlocksMemCache( + val maxSize: Int = 64 +) { + + private val mapping = ConcurrentHashMap>() + private val queue = ConcurrentLinkedQueue() + + fun get(hash: BlockHash): Mono> { + return Mono.justOrEmpty(mapping[hash]) + } + + fun add(block: BlockJson) { + mapping.put(block.hash, block) + queue.add(block.hash) + + while (queue.size > maxSize) { + val old = queue.remove() + mapping.remove(old) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt new file mode 100644 index 00000000..276e2baf --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt @@ -0,0 +1,27 @@ +package io.emeraldpay.dshackle.reader + +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Mono +import reactor.retry.Repeat +import java.time.Duration + +class BlockApiReader( + val upstream: Upstream +): Reader> { + + override fun read(key: BlockHash): Mono> { + return Mono.just(key) + .flatMap { + upstream.getApi(Selector.empty).executeAndConvert(Commands.eth().getBlock(it)) + }.repeatWhenEmpty { n -> + Repeat.times(3) + .exponentialBackoff(Duration.ofMillis(100), Duration.ofMillis(500)) + .apply(n) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt new file mode 100644 index 00000000..752c3626 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt @@ -0,0 +1,16 @@ +package io.emeraldpay.dshackle.reader + +import io.emeraldpay.dshackle.cache.BlocksMemCache +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Mono + +class BlockCacheReader( + val cache: BlocksMemCache +): Reader> { + + override fun read(key: BlockHash): Mono> { + return cache.get(key) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt new file mode 100644 index 00000000..8944dff4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt @@ -0,0 +1,23 @@ +package io.emeraldpay.dshackle.reader + +import reactor.core.publisher.Mono + +class CompoundReader( + private val readers: Collection> +): Reader { + + override fun read(key: K): Mono { + if (readers.isEmpty()) { + return Mono.empty() + } + var result = readers.first().read(key) + if (readers.size == 1) { + return result + } + readers.stream().skip(1).forEach { + result = result.switchIfEmpty(it.read(key)) + } + return result + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/EmptyReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/EmptyReader.kt new file mode 100644 index 00000000..77df0c6a --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/EmptyReader.kt @@ -0,0 +1,10 @@ +package io.emeraldpay.dshackle.reader + +import reactor.core.publisher.Mono + +class EmptyReader: Reader { + + override fun read(key: K): Mono { + return Mono.empty() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/Reader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/Reader.kt new file mode 100644 index 00000000..d5fb5752 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/Reader.kt @@ -0,0 +1,9 @@ +package io.emeraldpay.dshackle.reader + +import reactor.core.publisher.Mono + +interface Reader { + + fun read(key: K): Mono + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 855a930c..545504b8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -31,16 +31,16 @@ class NativeCall( return requestMono.flatMapMany(this::prepareCall) .map(this::setupCallParams) .parallel() - .flatMap(this::executeOnRemote) + .flatMap(this::fetch) .sequential() .map(this::buildResponse) .doOnError { e -> log.warn("Error during native call", e) } .onErrorResume(this::processException) } - fun setupCallParams(it: CallContext>): CallContext>> { - val params = extractParams(it.payload.t2) - return it.withPayload(Tuples.of(it.payload.t1, params)) + fun setupCallParams(it: CallContext): CallContext { + val params = extractParams(it.payload.params) + return it.withPayload(ParsedCallDetails(it.payload.method, params)) } fun buildResponse(it: CallContext): BlockchainOuterClass.NativeCallReplyItem { @@ -65,46 +65,57 @@ class NativeCall( .toMono() } - fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux>> { + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux> { val chain = Chain.byId(request.chain.number) if (chain == Chain.UNSPECIFIED) { - return Flux.error>>(CallFailure(0, Exception("Invalid chain id: ${request.chain.number}"))) + return Flux.error(CallFailure(0, Exception("Invalid chain id: ${request.chain.number}"))) } val upstream = upstreams.getUpstream(chain) - ?: return Flux.error>>(CallFailure(0, Exception("Chain ${chain.id} is unavailable"))) + ?: return Flux.error(CallFailure(0, Exception("Chain ${chain.id} is unavailable"))) return prepareCall(request, upstream) } - fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux>> { + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux> { val matcher = Selector.convertToMatcher(request.selector) - val apis = upstream.getApis(matcher) return request.itemsList.toFlux().map { val method = it.method val params = it.payload.toStringUtf8() val callQuorum = upstream.targets?.getQuorumFor(method) ?: AlwaysQuorum() callQuorum.init(upstream.getHead()) - CallContext(it.id, apis, callQuorum, Tuples.of(method, params)) + CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params)) } } - fun executeOnRemote(ctx: CallContext>>): Mono> { + fun fetch(ctx: CallContext): Mono> { + return fetchFromCache(ctx) + .switchIfEmpty( + Mono.just(ctx).flatMap(this::executeOnRemote) + ) + } + + fun fetchFromCache(ctx: CallContext): Mono> { + val cachingApi = ctx.upstream.cache + return cachingApi.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { ctx.withPayload(it) } + } + + fun executeOnRemote(ctx: CallContext): Mono> { val p: Predicate = CallQuorum.untilResolved(ctx.callQuorum) - val all = ctx.apis.toFlux().share() + val all = ctx.getApis().toFlux().share() //execute on the first API immediately, and then make a delay between each call to not dos upstreams val immediate = Flux.from(all).take(1) val retries = Flux.from(all).delayElements(Duration.ofMillis(200)) return Flux.concat(immediate, retries) .takeWhile(p) .flatMap { api -> - api.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { Tuples.of(it, api.upstream!!) } + api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) } } .reduce(ctx.callQuorum, CallQuorum.asReducer()) .filter { it.isResolved() } .map { val result = it.getResult() - ?: throw CallFailure(ctx.id, Exception("No response from upstream for ${ctx.payload.t1}")) + ?: throw CallFailure(ctx.id, Exception("No response from upstream for ${ctx.payload.method}")) ctx.withPayload(result) } .onErrorMap { @@ -113,7 +124,7 @@ class NativeCall( else CallFailure(ctx.id, it) } .switchIfEmpty( - Mono.error>(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.t1}"))) + Mono.error>(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.method}"))) ) } @@ -125,11 +136,18 @@ class NativeCall( return req as List } - open class CallContext(val id: Int, val apis: Iterator, val callQuorum: CallQuorum, val payload: T) { + open class CallContext(val id: Int, val upstream: AggregatedUpstream, val matcher: Selector.Matcher, val callQuorum: CallQuorum, val payload: T) { fun withPayload(payload: X): CallContext { - return CallContext(id, apis, callQuorum, payload) + return CallContext(id, upstream, matcher, callQuorum, payload) + } + + fun getApis(): Iterator { + return upstream.getApis(matcher) } } open class CallFailure(val id: Int, val reason: Throwable): Exception("Failed to call $id: ${reason.message}") + + class RawCallDetails(val method: String, val params: String) + class ParsedCallDetails(val method: String, val params: List) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index a75ef2ef..b6b1bdc4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -1,19 +1,40 @@ package io.emeraldpay.dshackle.upstream +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.cache.BlocksMemCache import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.BlockCacheReader +import io.emeraldpay.dshackle.reader.CompoundReader +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import org.springframework.context.Lifecycle +import reactor.core.Disposable import reactor.core.publisher.Flux import java.time.Duration import java.time.Instant import java.util.concurrent.atomic.AtomicReference +import java.util.concurrent.locks.ReentrantLock import java.util.function.Predicate +import kotlin.concurrent.withLock abstract class AggregatedUpstream( - val targets: CallMethods -): Upstream { + val targets: CallMethods, + val objectMapper: ObjectMapper +): Upstream, Lifecycle { + + private val blocksCache = BlocksMemCache() + private var cacheSubscription: Disposable? = null + private val blockReader: Reader> = CompoundReader( + listOf(BlockCacheReader(blocksCache)) + ) + var cache: CachingEthereumApi = CachingEthereumApi.empty() + private val reconfigLock = ReentrantLock() abstract fun getAll(): List abstract fun addUpstream(upstream: Upstream) - abstract fun getApis(matcher: Selector.Matcher): Iterator + abstract fun getApis(matcher: Selector.Matcher): Iterator override fun observeStatus(): Flux { val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } } @@ -61,4 +82,23 @@ abstract class AggregatedUpstream( return changed } } + + override fun start() { + } + + override fun stop() { + cacheSubscription?.dispose() + cacheSubscription = null + } + + fun onHeadUpdated(head: EthereumHead) { + reconfigLock.withLock { + cacheSubscription?.dispose() + cacheSubscription = head.getFlux().subscribe { + blocksCache.add(it) + } + cache = CachingEthereumApi(objectMapper, blockReader, head) + } + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt new file mode 100644 index 00000000..75cdb99e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt @@ -0,0 +1,48 @@ +package io.emeraldpay.dshackle.upstream + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.reader.EmptyReader +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.hex.HexQuantity +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.ResponseJson +import reactor.core.publisher.Mono +import java.util.function.Function + +open class CachingEthereumApi( + private val objectMapper: ObjectMapper, + private val cache: Reader>, + private val head: EthereumHead +): EthereumApi(objectMapper) { + + companion object { + @JvmStatic + fun empty(): CachingEthereumApi { + return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyEthereumHead()) + } + } + + override fun execute(id: Int, method: String, params: List): Mono { + return when (method) { + "eth_blockNumber" -> head.getFlux().next() + .map { HexQuantity.from(it.number).toHex() } + .map(toJson(id)) + "eth_getBlockByHash" -> Mono.just(params[0]) + .map { BlockHash.from(it as String) } + .flatMap(cache::read) + .map(toJson(id)) + else -> Mono.empty() + } + } + + fun toJson(id: Int): Function { + return Function { data -> + val resp = ResponseJson() + resp.id = id + resp.result = data + objectMapper.writer().writeValueAsBytes(resp) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index d6b193e9..5549f567 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -1,19 +1,19 @@ package io.emeraldpay.dshackle.upstream +import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable -import java.io.Closeable import java.lang.IllegalStateException import java.time.Duration -class ChainUpstreams ( +open class ChainUpstreams ( val chain: Chain, private val upstreams: MutableList, - targets: CallMethods -) : AggregatedUpstream(targets), Lifecycle { - + targets: CallMethods, + objectMapper: ObjectMapper +) : AggregatedUpstream(targets, objectMapper), Lifecycle { private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 @@ -30,12 +30,14 @@ class ChainUpstreams ( } override fun start() { + super.start() subscription = observeStatus() .distinctUntilChanged() .subscribe { printStatus() } } override fun stop() { + super.stop() subscription?.dispose() subscription = null head?.let { @@ -54,7 +56,7 @@ class ChainUpstreams ( } lagObserver?.stop() lagObserver = null - return if (upstreams.size == 1) { + val head = if (upstreams.size == 1) { val upstream = upstreams.first() upstream.setLag(0) upstream.getHead() @@ -68,6 +70,8 @@ class ChainUpstreams ( this.lagObserver = lagObserver newHead } + onHeadUpdated(head) + return head } override fun getAll(): List { @@ -79,7 +83,7 @@ class ChainUpstreams ( head = updateHead() } - override fun getApis(matcher: Selector.Matcher): Iterator { + override fun getApis(matcher: Selector.Matcher): Iterator { val i = seq++ if (seq >= Int.MAX_VALUE / 2) { seq = 0 @@ -87,7 +91,7 @@ class ChainUpstreams ( return FilteringApiIterator(upstreams, i, matcher) } - override fun getApi(matcher: Selector.Matcher): EthereumApi { + override fun getApi(matcher: Selector.Matcher): DirectEthereumApi { return getApis(matcher).next() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index adf6007d..0bc16ebc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -104,7 +104,7 @@ open class ConfiguredUpstreams( chain: Chain, options: UpstreamsConfig.Options, labels: UpstreamsConfig.Labels) { - var rpcApi: EthereumApi? = null + var rpcApi: DirectEthereumApi? = null val urls = ArrayList() up.rpc?.let { endpoint -> val rpcTransport = DefaultRpcTransport(endpoint.url) @@ -117,10 +117,9 @@ open class ConfiguredUpstreams( } } val rpcClient = DefaultRpcClient(rpcTransport) - rpcApi = EthereumApi( + rpcApi = DirectEthereumApi( rpcClient, objectMapper, - chain, targetFor(chain) ) urls.add(endpoint.url) @@ -171,7 +170,7 @@ open class ConfiguredUpstreams( override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams { val current = chainMapping[chain] if (current == null) { - val created = ChainUpstreams(chain, ArrayList(), targetFor(chain)) + val created = ChainUpstreams(chain, ArrayList(), targetFor(chain), objectMapper) created.addUpstream(up) created.start() chainMapping[chain] = created diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectEthereumApi.kt new file mode 100644 index 00000000..29d7efb8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectEthereumApi.kt @@ -0,0 +1,58 @@ +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.json.ResponseJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.time.Duration + +open class DirectEthereumApi( + val rpcClient: RpcClient, + private val objectMapper: ObjectMapper, + val targets: CallMethods +): EthereumApi(objectMapper) { + + private val timeout = Duration.ofSeconds(5) + private val log = LoggerFactory.getLogger(EthereumApi::class.java) + + override fun execute(id: Int, method: String, params: List): Mono { + val result: Mono = when { + targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) } + targets.isAllowed(method) -> callUpstream(method, params) + else -> Mono.error(RpcException(-32601, "Method not allowed or not found")) + } + return result + .doOnError { t -> + log.warn("Upstream error: ${t.message} for ${method}") + } + .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)) + } + } + + 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"))) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyEthereumHead.kt new file mode 100644 index 00000000..4771d117 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyEthereumHead.kt @@ -0,0 +1,12 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Flux + +class EmptyEthereumHead : EthereumHead { + + override fun getFlux(): Flux> { + return Flux.empty() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index bce365ca..f25a16b2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -7,28 +7,22 @@ import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.ResponseJson import org.slf4j.LoggerFactory import reactor.core.publisher.Mono +import java.io.InputStream import java.time.Duration -open class EthereumApi( - val rpcClient: RpcClient, - private val objectMapper: ObjectMapper, - private val chain: Chain, - val targets: CallMethods +abstract class EthereumApi( + objectMapper: ObjectMapper ) { private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) var upstream: Upstream? = null - private val timeout = Duration.ofSeconds(5) - private val log = LoggerFactory.getLogger(EthereumApi::class.java) - var ws: EthereumWs? = null - set(value) { - field = value - } + abstract fun execute(id: Int, method: String, params: List): Mono - open fun executeAndConvert(rpcCall: RpcCall): Mono { + fun executeAndConvert(rpcCall: RpcCall): Mono { val convertToJS = java.util.function.Function> { resp -> - val jsonValue: JS? = jacksonRpcConverter.fromJson(resp.inputStream(), rpcCall.jsonType, Int::class.java) + val inputStream: InputStream = resp.inputStream() + val jsonValue: JS? = jacksonRpcConverter.fromJson(inputStream, rpcCall.jsonType, Int::class.java) if (jsonValue == null) Mono.empty() else Mono.just(jsonValue) } @@ -36,51 +30,4 @@ open class EthereumApi( .flatMap(convertToJS) .map(rpcCall.converter::apply) } - - open fun execute(id: Int, method: String, params: List): Mono { - val result: Mono = when { - targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) } - targets.isAllowed(method) -> callUpstream(method, params) - else -> Mono.error(RpcException(-32601, "Method not allowed or not found")) - } - return result - .doOnError { t -> - log.warn("Upstream error: ${t.message} for ${method} on $chain") - } - .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)) - } - } - - private fun callUpstream(method: String, params: List): Mono { - if (method == "eth_blockNumber") { - val current = upstream?.getHead()?.getFlux()?.next()?.let { head -> - head.map { HexQuantity.from(it.number).toHex() } - } - if (current != null) { - return current - } - } - return Mono.fromCompletionStage( - 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/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt index 99edc820..0409cd4f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt @@ -14,7 +14,7 @@ import java.time.Duration import java.util.concurrent.atomic.AtomicReference class EthereumRpcHead( - private val api: EthereumApi + private val api: DirectEthereumApi ): EthereumHead, Lifecycle { private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index b15110ef..2e4e8757 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -8,14 +8,14 @@ import reactor.core.Disposable open class EthereumUpstream( val chain: Chain, - private val api: EthereumApi, + private val api: DirectEthereumApi, private val ethereumWs: EthereumWs? = null, private val options: UpstreamsConfig.Options, val node: NodeDetailsList.NodeDetails, private val targets: CallMethods ): DefaultUpstream(), Lifecycle { - constructor(chain: Chain, api: EthereumApi): this(chain, api, null, + constructor(chain: Chain, api: DirectEthereumApi): this(chain, api, null, UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()), DirectCallMethods()) @@ -84,11 +84,11 @@ open class EthereumUpstream( return head } - override fun getApi(matcher: Selector.Matcher): EthereumApi { + override fun getApi(matcher: Selector.Matcher): DirectEthereumApi { return api } - fun getApi(): EthereumApi { + fun getApi(): DirectEthereumApi { return api } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt index 7a1caeef..f7483ab3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt @@ -5,7 +5,7 @@ class FilteringApiIterator( private var pos: Int, private val matcher: Selector.Matcher, private val repeatLimit: Int = 3 -): Iterator { +): Iterator { private var nextUpstream: Upstream? = null private var consumed = 0 @@ -31,7 +31,7 @@ class FilteringApiIterator( return nextInternal() } - override fun next(): EthereumApi { + override fun next(): DirectEthereumApi { if (nextInternal()) { val curr = nextUpstream!! nextUpstream = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index 34901960..71df82a6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -44,9 +44,9 @@ open class GrpcUpstream( private var headSubscription: Disposable? = null - open fun createApi(matcher: Selector.Matcher): EthereumApi { + open fun createApi(matcher: Selector.Matcher): DirectEthereumApi { val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher)) - return EthereumApi(rpcClient, objectMapper, chain, targets).let { + return DirectEthereumApi(rpcClient, objectMapper, targets).let { it.upstream = this it } @@ -156,7 +156,7 @@ open class GrpcUpstream( return head } - override fun getApi(matcher: Selector.Matcher): EthereumApi { + override fun getApi(matcher: Selector.Matcher): DirectEthereumApi { return createApi(matcher) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index fcaf70f2..c79bf851 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -8,7 +8,8 @@ interface Upstream { fun getStatus(): UpstreamAvailability fun observeStatus(): Flux fun getHead(): EthereumHead - fun getApi(matcher: Selector.Matcher): EthereumApi + fun getApi(matcher: Selector.Matcher): DirectEthereumApi +// fun getCache(): CachingEthereumApi fun getOptions(): UpstreamsConfig.Options fun getSupportedTargets(): Set fun setLag(lag: Long) diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy new file mode 100644 index 00000000..d58d71b1 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy @@ -0,0 +1,52 @@ +package io.emeraldpay.dshackle.cache + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import spock.lang.Specification + +class BlocksMemCacheSpec extends Specification { + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" + String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" + + def "Add and read"() { + setup: + def cache = new BlocksMemCache() + def block = new BlockJson() + block.number = 100 + block.hash = BlockHash.from(hash1) + + when: + cache.add(block) + def act = cache.get(BlockHash.from(hash1)).block() + then: + act == block + } + + def "Keeps only configured amount"() { + setup: + def cache = new BlocksMemCache(3) + [hash1] + + when: + [hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i -> + def block = new BlockJson() + block.number = 100 + i + block.hash = BlockHash.from(hash) + cache.add(block) + } + + def act1 = cache.get(BlockHash.from(hash1)).block() + def act2 = cache.get(BlockHash.from(hash2)).block() + def act3 = cache.get(BlockHash.from(hash3)).block() + def act4 = cache.get(BlockHash.from(hash4)).block() + then: + act2.hash.toHex() == hash2 + act3.hash.toHex() == hash3 + act4.hash.toHex() == hash4 + act1 == null + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 45f2b8e0..fa222f83 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -1,17 +1,23 @@ package io.emeraldpay.dshackle.rpc +import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.test.EthereumApiMock import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.AlwaysQuorum +import io.emeraldpay.dshackle.upstream.CachingEthereumApi import io.emeraldpay.dshackle.upstream.CallQuorum +import io.emeraldpay.dshackle.upstream.DirectEthereumApi import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.dshackle.upstream.NonEmptyQuorum +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 reactor.core.publisher.Mono import reactor.test.StepVerifier import reactor.util.function.Tuples import spock.lang.Specification @@ -33,7 +39,9 @@ class NativeCallSpec extends Specification { apiMock.answer("eth_test", [], "foo") def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, [apiMock].multiply(59).iterator(), quorum, Tuples.of("eth_test", [])) + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), + Selector.empty, quorum, + new NativeCall.ParsedCallDetails("eth_test", [])) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) @@ -59,7 +67,9 @@ class NativeCallSpec extends Specification { apiMock.answerOnce("eth_test", [], null) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", [])) + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), + Selector.empty, quorum, + new NativeCall.ParsedCallDetails("eth_test", [])) when: @@ -85,7 +95,8 @@ class NativeCallSpec extends Specification { apiMock.answerOnce("eth_test", [], "foo") def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", [])) + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), Selector.empty, quorum, + new NativeCall.ParsedCallDetails("eth_test", [])) (4..5) * quorum.isResolved() 3 * quorum.record(_, _) @@ -135,9 +146,10 @@ class NativeCallSpec extends Specification { def upstreams = Stub(Upstreams) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def json = [jsonrpc:"2.0", id:1, result: "foo"] + when: def resp = nativeCall.buildResponse( - new NativeCall.CallContext(1561, [].iterator(), new AlwaysQuorum(), objectMapper.writeValueAsBytes(json)) + new NativeCall.CallContext(1561, TestingCommons.aggregatedUpstream(Stub(DirectEthereumApi)), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json)) ) then: resp.id == 1561 @@ -191,4 +203,42 @@ class NativeCallSpec extends Specification { // .expectComplete() .verify(Duration.ofSeconds(1)) } + + def "Calls cache before remote"() { + setup: + def upstreams = Stub(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def api = Mock(DirectEthereumApi) + def upstream = TestingCommons.aggregatedUpstream(api) + def cacheMock = Mock(CachingEthereumApi) + upstream.cache = cacheMock + + def ctx = new NativeCall.CallContext(10, + upstream, + Selector.empty, new AlwaysQuorum(), + new NativeCall.ParsedCallDetails("eth_test", [])) + when: + nativeCall.fetch(ctx) + then: + 1 * cacheMock.execute(10, "eth_test", []) >> Mono.empty() + } + + def "Uses cached value"() { + setup: + def upstreams = Stub(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def upstream = TestingCommons.aggregatedUpstream(Stub(DirectEthereumApi)) + def cacheMock = Mock(CachingEthereumApi) + upstream.cache = cacheMock + + def ctx = new NativeCall.CallContext(10, + upstream, + Selector.empty, new AlwaysQuorum(), + new NativeCall.ParsedCallDetails("eth_test", [])) + when: + def act = nativeCall.fetch(ctx) + then: + 1 * cacheMock.execute(10, "eth_test", []) >> Mono.just('{"result": "foo"}'.bytes) + new String(act.block().payload) == '{"result": "foo"}' + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy index f326847c..6254a282 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy @@ -5,6 +5,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.UpstreamsMock +import io.emeraldpay.dshackle.upstream.DirectEthereumApi import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.grpc.Chain @@ -57,7 +58,7 @@ class StreamHeadSpec extends Specification { .build() } - def upstream = new EthereumUpstreamMock(Chain.ETHEREUM, Mock(EthereumApi)) + def upstream = new EthereumUpstreamMock(Chain.ETHEREUM, Stub(DirectEthereumApi.class)) def upstreams = new UpstreamsMock(Chain.ETHEREUM, upstream) def streamHead = new StreamHead(upstreams) when: diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index 3eb615e6..43e2a3c8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.DirectCallMethods +import io.emeraldpay.dshackle.upstream.DirectEthereumApi import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.dshackle.upstream.QuorumBasedMethods import io.emeraldpay.dshackle.upstream.Upstream @@ -17,14 +18,14 @@ import org.slf4j.Logger import org.slf4j.LoggerFactory import reactor.core.publisher.Mono -class EthereumApiMock extends EthereumApi { +class EthereumApiMock extends DirectEthereumApi { private static final Logger log = LoggerFactory.getLogger(this) List predefined = [] private ObjectMapper objectMapper EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) { - super(rpcClient, objectMapper, chain, new DirectCallMethods()) + super(rpcClient, objectMapper, new DirectCallMethods()) this.objectMapper = objectMapper } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 7e650cde..000edacb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.test +import io.emeraldpay.dshackle.upstream.DirectEthereumApi import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.dshackle.upstream.EthereumHead import io.emeraldpay.dshackle.upstream.EthereumUpstream @@ -13,7 +14,7 @@ class EthereumUpstreamMock extends EthereumUpstream { EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() - EthereumUpstreamMock(@NotNull Chain chain, @NotNull EthereumApi api) { + EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api) { super(chain, api) setLag(0) setStatus(UpstreamAvailability.OK) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 21aa8ebb..c1d80a16 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -4,6 +4,10 @@ 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.upstream.AggregatedUpstream +import io.emeraldpay.dshackle.upstream.ChainUpstreams +import io.emeraldpay.dshackle.upstream.DirectCallMethods +import io.emeraldpay.dshackle.upstream.DirectEthereumApi import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.dshackle.upstream.EthereumUpstream import io.emeraldpay.dshackle.upstream.Upstream @@ -42,7 +46,11 @@ class TestingCommons { return new JacksonRpcConverter(objectMapper()) } - static EthereumUpstreamMock upstream(EthereumApi api) { + static EthereumUpstreamMock upstream(DirectEthereumApi api) { return new EthereumUpstreamMock(Chain.ETHEREUM, api) } + + static AggregatedUpstream aggregatedUpstream(DirectEthereumApi api) { + return new ChainUpstreams(Chain.ETHEREUM, [upstream(api)], new DirectCallMethods(), objectMapper()) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy index 1ea2bb9d..23f65a53 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy @@ -26,7 +26,7 @@ class UpstreamsMock implements Upstreams { @Override AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { if (!upstreams.containsKey(chain)) { - upstreams[chain] = new ChainUpstreams(chain, [up], targetFor(chain)) + upstreams[chain] = new ChainUpstreams(chain, [up], targetFor(chain), TestingCommons.objectMapper()) } else { upstreams[chain].addUpstream(up) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index 8cdc53ec..22f93a5a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -23,11 +23,13 @@ class EthereumGrpcTransportSpec extends Specification { def "Make simple call"() { setup: + def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) + def callData = [:] def otherSideUpstreams = Mock(Upstreams) - def otherSideAggr = Mock(AggregatedUpstream) + def otherSideAggr = TestingCommons.aggregatedUpstream(otherSideApi) + def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) - def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) otherSideApi.upstream = otherSideAggr def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @@ -47,9 +49,6 @@ class EthereumGrpcTransportSpec extends Specification { then: 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr - 1 * otherSideAggr.getApis(_) >> [otherSideApi].iterator() - _ * otherSideAggr.getHead() >> Stub(EthereumHead) - _ * otherSideAggr.getTargets() >> ethereumTargets status.failed == 0 status.succeed == 1 status.total == 1 @@ -67,11 +66,12 @@ class EthereumGrpcTransportSpec extends Specification { def "Make few calls"() { setup: + def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) + def callData = [:] def otherSideUpstreams = Mock(Upstreams) - def otherSideAggr = Mock(AggregatedUpstream) + def otherSideAggr = TestingCommons.aggregatedUpstream(otherSideApi) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) - def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) otherSideApi.upstream = otherSideAggr def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @@ -95,9 +95,6 @@ class EthereumGrpcTransportSpec extends Specification { then: 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr - 1 * otherSideAggr.getApis(_) >> [otherSideApi].multiply(34).iterator() - _ * otherSideAggr.getHead() >> Stub(EthereumHead) - _ * otherSideAggr.getTargets() >> ethereumTargets status.failed == 0 status.succeed == 2 status.total == 2 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy index a2a519f4..bfaec6ef 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy @@ -23,7 +23,7 @@ class FilteringApiIteratorSpec extends Specification { ].collect { new EthereumUpstream( Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM, ethereumTargets), + new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets), (EthereumWs) null, new UpstreamsConfig.Options(), new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)),