diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index d409ea1e..7f364335 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -5,7 +5,6 @@ import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.RpcException import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -77,7 +76,7 @@ class NativeCall( return prepareCall(request, upstream) } - fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstreams): 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 { @@ -109,6 +108,7 @@ class NativeCall( ctx.withPayload(result) } .onErrorMap { + log.error("Failed to make a call", it) if (it is CallFailure) it else CallFailure(ctx.id, it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 2f46522a..a1d12685 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -3,8 +3,6 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.upstream.AvailableChains -import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.UpstreamServices import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain @@ -17,16 +15,13 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux -import java.lang.Exception -import java.time.Duration import java.util.concurrent.ConcurrentLinkedQueue import javax.annotation.PostConstruct import kotlin.collections.HashMap @Service class StreamHead( - @Autowired private val upstreams: Upstreams, - @Autowired private val availableChains: AvailableChains + @Autowired private val upstreams: Upstreams ) { private val log = LoggerFactory.getLogger(StreamHead::class.java) @@ -34,14 +29,17 @@ class StreamHead( @PostConstruct fun init() { - availableChains.observe().subscribe { chain -> + upstreams.observeChains().subscribe { chain -> + if (clients.containsKey(chain)) { + return@subscribe + } clients[chain] = ConcurrentLinkedQueue() subscribe(chain) } } private fun subscribe(chain: Chain) { - upstreams.getUpstream(chain)?.let { up -> + upstreams.getUpstream(chain)!!.let { up -> up.getHead() .getFlux() .doOnComplete { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index e8665c47..0f910e2f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -11,13 +11,12 @@ import reactor.core.publisher.Mono @Service class SubscribeStatus( - @Autowired private val upstreams: Upstreams, - @Autowired private val availableChains: AvailableChains + @Autowired private val upstreams: Upstreams ) { fun subscribeStatus(requestMono: Mono): Flux { return requestMono.flatMapMany { - val ups = availableChains.getAll().mapNotNull { chain -> + val ups = upstreams.getAvailable().mapNotNull { chain -> val chainUpstream = upstreams.getUpstream(chain) chainUpstream?.observeStatus()?.map { avail -> ChainSubscription(chain, chainUpstream, avail) @@ -45,6 +44,6 @@ class SubscribeStatus( .build() } - class ChainSubscription(val chain: Chain, val up: AggregatedUpstreams, val avail: UpstreamAvailability) + class ChainSubscription(val chain: Chain, val up: AggregatedUpstream, val avail: UpstreamAvailability) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index 0ca30803..44adda38 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -2,7 +2,6 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.upstream.AvailableChains import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain @@ -30,7 +29,6 @@ import javax.annotation.PostConstruct @Service class TrackAddress( @Autowired private val upstreams: Upstreams, - @Autowired private val availableChains: AvailableChains, @Autowired private val upstreamScheduler: Scheduler ) { @@ -40,7 +38,7 @@ class TrackAddress( @PostConstruct fun init() { - availableChains.observe().subscribe { chain -> + upstreams.observeChains().subscribe { chain -> if (!clients.containsKey(chain)) { clients[chain] = ConcurrentLinkedQueue() upstreams.getUpstream(chain)?.getHead()?.let { head -> @@ -53,7 +51,7 @@ class TrackAddress( @Scheduled(fixedDelay = 120_000) fun pingOld() { val period = Duration.ofMinutes(15) - availableChains.getAll().forEach { chain -> + upstreams.getAvailable().forEach { chain -> clients[chain]?.let { clients -> clients.toFlux().filter { it.lastPing < Instant.now().minus(period) @@ -80,7 +78,7 @@ class TrackAddress( private fun initializeSimple(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) - if (!availableChains.supports(chain)) { + if (!upstreams.isAvailable(chain)) { return Flux.error(Exception("Unsupported chain ${request.asset.chainValue}")) } if (request.asset.code?.toLowerCase() != "ether") { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index a63f84ec..2f52e0f3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -3,7 +3,6 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.upstream.AvailableChains import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams @@ -38,7 +37,6 @@ import kotlin.math.min @Service class TrackTx( @Autowired private val upstreams: Upstreams, - @Autowired private val availableChains: AvailableChains, @Autowired private val upstreamScheduler: Scheduler ) { @@ -59,7 +57,7 @@ class TrackTx( @PostConstruct fun init() { - availableChains.observe().subscribe { chain -> + upstreams.observeChains().subscribe { chain -> clients[chain] = ConcurrentLinkedQueue() upstreams.getUpstream(chain)?.getHead()?.let { head -> head.getFlux().subscribe { verifyAll(chain) } @@ -239,6 +237,9 @@ class TrackTx( .executeAndConvert(Commands.eth().getTransaction(tx.txid)) return execution .flatMap { updateFromBlock(upstream, tx, it) } + .doOnError { t -> + log.error("Failed to load tx block", t) + } .switchIfEmpty(Mono.just(tx.withStatus(found = false))) .filter { current -> initialStatus != current.status || current.shouldNotify() || current.shouldClose() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt similarity index 96% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index d03efdb5..a75ef2ef 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -7,8 +7,8 @@ import java.time.Instant import java.util.concurrent.atomic.AtomicReference import java.util.function.Predicate -abstract class AggregatedUpstreams( - val targets: EthereumTargets +abstract class AggregatedUpstream( + val targets: CallMethods ): Upstream { abstract fun getAll(): List diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt deleted file mode 100644 index 3fcee018..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt +++ /dev/null @@ -1,52 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.grpc.Chain -import org.springframework.beans.factory.annotation.Autowired -import org.springframework.stereotype.Repository -import reactor.core.publisher.Flux -import reactor.core.publisher.TopicProcessor -import java.util.* -import kotlin.collections.LinkedHashSet - -@Repository -class AvailableChains( - @Autowired private val objectMapper: ObjectMapper -) { - - private val all = LinkedHashSet() - private val bus = TopicProcessor.create() - private val callTargets = HashMap() - - fun add(chain: Chain) { - if (all.contains(chain)) { - return - } - all.add(chain) - bus.onNext(chain) - } - - fun observe(): Flux { - return Flux.merge( - Flux.fromIterable(all), - Flux.from(bus) - ) - } - - fun supports(chain: Chain): Boolean { - return all.contains(chain) - } - - fun getAll(): Set { - return Collections.unmodifiableSet(all) - } - - fun targetFor(chain: Chain): EthereumTargets { - var current = callTargets[chain] - if (current == null) { - current = EthereumTargets(objectMapper, chain) - callTargets[chain] = current - } - return current - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallMethods.kt new file mode 100644 index 00000000..912aac0c --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallMethods.kt @@ -0,0 +1,9 @@ +package io.emeraldpay.dshackle.upstream + +interface CallMethods { + fun getQuorumFor(method: String): CallQuorum + fun isAllowed(method: String): Boolean + fun getSupportedMethods(): Set + fun isHardcoded(method: String): Boolean + fun hardcoded(method: String): Any +} \ 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 39172278..cffbdf2c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -9,8 +9,8 @@ import java.time.Duration class ChainUpstreams ( val chain: Chain, private val upstreams: MutableList, - targets: EthereumTargets -) : AggregatedUpstreams(targets) { + targets: CallMethods +) : AggregatedUpstream(targets) { private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index f916aade..e0e0e114 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -12,6 +12,8 @@ import org.springframework.beans.factory.annotation.Autowired import org.springframework.core.env.Environment import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Repository +import reactor.core.publisher.Flux +import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import java.io.File import java.net.URI @@ -23,12 +25,13 @@ import kotlin.collections.HashMap @Repository open class ConfiguredUpstreams( @Autowired val env: Environment, - @Autowired private val objectMapper: ObjectMapper, - @Autowired private val availableChains: AvailableChains + @Autowired private val objectMapper: ObjectMapper ) : Upstreams { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) private val chainMapping = ConcurrentHashMap() + private val chainsBus = TopicProcessor.create() + private val callTargets = HashMap() private val chainNames = mapOf( "ethereum" to Chain.ETHEREUM, @@ -108,7 +111,7 @@ open class ConfiguredUpstreams( rpcClient, objectMapper, chain, - availableChains.targetFor(chain) + targetFor(chain) ) urls.add(endpoint.url) } @@ -122,12 +125,7 @@ open class ConfiguredUpstreams( } if (rpcApi != null) { log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - getOrCreateUpstream(chain) - .addUpstream( - EthereumUpstream( - chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), availableChains.targetFor(chain) - ) - ) + addUpstream(chain, EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain))) } } @@ -139,7 +137,7 @@ open class ConfiguredUpstreams( objectMapper, options, up.auth, - availableChains + this ) log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") ds.start() @@ -148,21 +146,24 @@ open class ConfiguredUpstreams( } .subscribe { log.info("Subscribed to $it through gRPC at ${endpoint.host}:${endpoint.port}") - getOrCreateUpstream(it).addUpstream(ds.getOrCreate(it)) + addUpstream(it, ds.getOrCreate(it)) } } - override fun getUpstream(chain: Chain): AggregatedUpstreams? { + override fun getUpstream(chain: Chain): AggregatedUpstream? { return chainMapping[chain] } - override fun getOrCreateUpstream(chain: Chain): ChainUpstreams { + override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams { val current = chainMapping[chain] if (current == null) { - availableChains.add(chain) - val created = ChainUpstreams(chain, ArrayList(), availableChains.targetFor(chain)) + val created = ChainUpstreams(chain, ArrayList(), targetFor(chain)) + created.addUpstream(up) chainMapping[chain] = created + chainsBus.onNext(chain) return created + } else { + current.addUpstream(up) } return current } @@ -175,4 +176,24 @@ open class ConfiguredUpstreams( override fun getAvailable(): List { return Collections.unmodifiableList(chainMapping.keys.toList()) } + + override fun observeChains(): Flux { + return Flux.merge( + Flux.fromIterable(getAvailable()), + Flux.from(chainsBus) + ) + } + + override fun targetFor(chain: Chain): CallMethods { + var current = callTargets[chain] + if (current == null) { + current = QuorumBasedMethods(objectMapper, chain) + callTargets[chain] = current + } + return current + } + + override fun isAvailable(chain: Chain): Boolean { + return chainMapping.containsKey(chain) && callTargets.containsKey(chain) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 1ce755b7..221f85af 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -5,13 +5,13 @@ import reactor.core.publisher.TopicProcessor import java.util.concurrent.atomic.AtomicReference abstract class DefaultUpstream( - lag: Long, - avail: UpstreamAvailability + defaultLag: Long, + defaultAvail: UpstreamAvailability ) : Upstream { constructor() : this(Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE) - private val status = AtomicReference(Status(lag, avail, statusByLag(lag, avail))) + private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail))) private val statusStream: TopicProcessor = TopicProcessor.create() override fun getStatus(): UpstreamAvailability { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectCallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectCallMethods.kt new file mode 100644 index 00000000..e8dbe3ce --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectCallMethods.kt @@ -0,0 +1,24 @@ +package io.emeraldpay.dshackle.upstream + +class DirectCallMethods : CallMethods { + + override fun getQuorumFor(method: String): CallQuorum { + return AlwaysQuorum() + } + + override fun isAllowed(method: String): Boolean { + return true + } + + override fun getSupportedMethods(): Set { + return emptySet() + } + + override fun isHardcoded(method: String): Boolean { + return false + } + + override fun hardcoded(method: String): Any { + return "unsupported" + } +} \ 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 6ed89f55..043e9348 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -13,11 +13,11 @@ open class EthereumApi( val rpcClient: RpcClient, private val objectMapper: ObjectMapper, private val chain: Chain, - val targets: EthereumTargets, - var upstream: Upstream? = null + val targets: CallMethods ) { private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) + var upstream: Upstream? = null private val timeout = Duration.ofSeconds(5) private val log = LoggerFactory.getLogger(EthereumApi::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 a297c7e9..50e94a37 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -2,13 +2,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory -import reactor.core.publisher.Flux -import reactor.core.publisher.TopicProcessor -import java.util.concurrent.atomic.AtomicLong -import java.util.concurrent.atomic.AtomicReference open class EthereumUpstream( val chain: Chain, @@ -16,22 +10,20 @@ open class EthereumUpstream( private val ethereumWs: EthereumWs? = null, private val options: UpstreamsConfig.Options, val node: NodeDetailsList.NodeDetails, - private val targets: EthereumTargets + private val targets: CallMethods ): DefaultUpstream() { + constructor(chain: Chain, api: EthereumApi): this(chain, api, null, + UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()), + DirectCallMethods()) + override fun getSupportedTargets(): Set { return targets.getSupportedMethods() } private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) - private val head: EthereumHead = if (ethereumWs != null) { - EthereumWsHead(ethereumWs) - } else { - EthereumRpcHead(api).apply { - this.start() - } - } + private val head: EthereumHead = createHead() private val validator = UpstreamValidator(this, options) @@ -43,6 +35,16 @@ open class EthereumUpstream( .subscribe(this::setStatus) } + open fun createHead(): EthereumHead { + return if (ethereumWs != null) { + EthereumWsHead(ethereumWs) + } else { + EthereumRpcHead(api).apply { + this.start() + } + } + } + override fun isAvailable(matcher: Selector.Matcher): Boolean { return getStatus() == UpstreamAvailability.OK && matcher.matches(node.labels) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt index 423a14e6..7a1caeef 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt @@ -1,30 +1,30 @@ package io.emeraldpay.dshackle.upstream class FilteringApiIterator( - private val apis: List, + private val upstreams: List, private var pos: Int, private val matcher: Selector.Matcher, private val repeatLimit: Int = 3 ): Iterator { - private var nextApi: Upstream? = null + private var nextUpstream: Upstream? = null private var consumed = 0 private fun nextInternal(): Boolean { - if (nextApi != null) { + if (nextUpstream != null) { return true } - while (nextApi == null) { + while (nextUpstream == null) { consumed++ - if (consumed > apis.size * repeatLimit) { + if (consumed > upstreams.size * repeatLimit) { return false } - val api = apis[pos++ % apis.size] - if (api.isAvailable(matcher)) { - nextApi = api + val upstream = upstreams[pos++ % upstreams.size] + if (upstream.isAvailable(matcher)) { + nextUpstream = upstream } } - return nextApi != null + return nextUpstream != null } override fun hasNext(): Boolean { @@ -33,8 +33,8 @@ class FilteringApiIterator( override fun next(): EthereumApi { if (nextInternal()) { - val curr = nextApi!! - nextApi = null + val curr = nextUpstream!! + nextUpstream = null return curr.getApi(matcher) } throw IllegalStateException("No upstream API available") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index d406bb2f..da3a785c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -11,7 +11,6 @@ import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.BlockTag import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -29,10 +28,10 @@ open class GrpcUpstream( private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, private val options: UpstreamsConfig.Options, - private val targets: EthereumTargets + private val targets: CallMethods ): DefaultUpstream() { - constructor(chain: Chain, client: ReactorBlockchainGrpc.ReactorBlockchainStub, objectMapper: ObjectMapper, targets: EthereumTargets) + constructor(chain: Chain, client: ReactorBlockchainGrpc.ReactorBlockchainStub, objectMapper: ObjectMapper, targets: CallMethods) : this(chain, client, objectMapper, UpstreamsConfig.Options.getDefaults(), targets) private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) @@ -48,7 +47,10 @@ open class GrpcUpstream( open fun createApi(matcher: Selector.Matcher): EthereumApi { val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher)) - return EthereumApi(rpcClient, objectMapper, chain, targets, this) + return EthereumApi(rpcClient, objectMapper, chain, targets).let { + it.upstream = this + it + } } open fun connect() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt index 243b1da5..0de7505b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt @@ -22,7 +22,7 @@ class GrpcUpstreams( private val objectMapper: ObjectMapper, private val options: UpstreamsConfig.Options, private val auth: UpstreamsConfig.TlsAuth? = null, - private val availableChains: AvailableChains + private val upstreams: Upstreams ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) @@ -92,9 +92,9 @@ class GrpcUpstreams( lock.withLock { val current = known[chain] return if (current == null) { - val created = GrpcUpstream(chain, client!!, objectMapper, options, availableChains.targetFor(chain)) + val created = GrpcUpstream(chain, client!!, objectMapper, options, upstreams.targetFor(chain)) known[chain] = created - availableChains.add(chain) + upstreams.addUpstream(chain, created) created.connect() created } else { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt similarity index 92% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt index 27603006..d87af109 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt @@ -6,10 +6,10 @@ import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.RpcException import java.util.* -class EthereumTargets( +class QuorumBasedMethods( private val objectMapper: ObjectMapper, private val chain: Chain -) { +) : CallMethods { private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) @@ -61,7 +61,7 @@ class EthereumTargets( "eth_accounts" ) - open fun getQuorumFor(method: String): CallQuorum { + override fun getQuorumFor(method: String): CallQuorum { return when { hardcodedMethods.contains(method) -> AlwaysQuorum() anyResponseMethods.contains(method) -> NotLaggingQuorum(6) @@ -78,14 +78,14 @@ class EthereumTargets( } } - fun isAllowed(method: String): Boolean { + override fun isAllowed(method: String): Boolean { return allowedMethods.contains(method) } - fun isHardcoded(method: String): Boolean { + override fun isHardcoded(method: String): Boolean { return hardcodedMethods.contains(method) } - fun hardcoded(method: String): Any { + override fun hardcoded(method: String): Any { if ("net_version" == method) { if (Chain.ETHEREUM == chain) { return "1" @@ -131,7 +131,7 @@ class EthereumTargets( throw RpcException(-32601, "Method not found") } - fun getSupportedMethods(): Set { + override fun getSupportedMethods(): Set { return allowedMethods.plus(hardcodedMethods).toSortedSet() } } \ 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 6977eb43..5b245eea 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -4,7 +4,10 @@ import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux interface Upstreams { - fun getOrCreateUpstream(chain: Chain): AggregatedUpstreams - fun getUpstream(chain: Chain): AggregatedUpstreams? + fun addUpstream(chain: Chain, up: Upstream): AggregatedUpstream + fun getUpstream(chain: Chain): AggregatedUpstream? fun getAvailable(): List + fun observeChains(): Flux + fun targetFor(chain: Chain): CallMethods + fun isAvailable(chain: Chain): Boolean } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 3f621a7d..025190cb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -27,14 +27,13 @@ class NativeCallSpec extends Specification { def quorum = Spy(new AlwaysQuorum()) def upstreams = Stub(Upstreams) RpcClient rpcClient = Stub(RpcClient) - def upstream = Stub(Upstream) - def apiMock = TestingCommons.api(rpcClient, upstream) + def apiMock = TestingCommons.api(rpcClient) + apiMock.upstream = Stub(Upstream) apiMock.answer("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, [apiMock].multiply(59).iterator(), quorum, Tuples.of("eth_test", [])) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) @@ -52,8 +51,8 @@ class NativeCallSpec extends Specification { def upstreams = Stub(Upstreams) RpcClient rpcClient = Stub(RpcClient) - def upstream = Stub(Upstream) - def apiMock = TestingCommons.api(rpcClient, upstream) + def apiMock = TestingCommons.api(rpcClient) + apiMock.upstream = Stub(Upstream) apiMock.answerOnce("eth_test", [], null) apiMock.answerOnce("eth_test", [], "bar") @@ -79,8 +78,8 @@ class NativeCallSpec extends Specification { def upstreams = Stub(Upstreams) RpcClient rpcClient = Stub(RpcClient) - def upstream = Stub(Upstream) - def apiMock = TestingCommons.api(rpcClient, upstream) + def apiMock = TestingCommons.api(rpcClient) + apiMock.upstream = Stub(Upstream) apiMock.answer("eth_test", [], null, 3) apiMock.answerOnce("eth_test", [], "foo") diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy index da4d9f1d..f5260a9b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy @@ -2,11 +2,8 @@ package io.emeraldpay.dshackle.rpc 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.AggregatedUpstreams -import io.emeraldpay.dshackle.upstream.AvailableChains -import io.emeraldpay.dshackle.upstream.EthereumHead +import io.emeraldpay.dshackle.test.UpstreamsMock import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address @@ -23,10 +20,6 @@ import java.time.Duration class TrackAddressSpec extends Specification { - AvailableChains availableChains - Upstreams upstreams - TrackAddress trackAddress - def chain = Common.ChainRef.CHAIN_ETHEREUM def address1 = "0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f" def address1Proto = Common.SingleAddress.newBuilder() @@ -36,18 +29,6 @@ class TrackAddressSpec extends Specification { .setCode("ETHER") - def setup() { - availableChains = new AvailableChains(TestingCommons.objectMapper()) - upstreams = Mock(Upstreams) - trackAddress = new TrackAddress(upstreams, availableChains, Schedulers.immediate()) - } - - def start() { - trackAddress.init() - availableChains.add(Chain.ETHEREUM) - availableChains.add(Chain.TESTNET_KOVAN) - } - def "get balance"() { setup: def req = BlockchainOuterClass.BalanceRequest.newBuilder() @@ -60,12 +41,13 @@ class TrackAddressSpec extends Specification { .setBalance("1234567890") .build() - def upstreamMock = Mock(AggregatedUpstreams) - def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackAddress trackAddress = new TrackAddress(upstreams, Schedulers.immediate()) + trackAddress.init() + apiMock.answer("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") - _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock - _ * upstreamMock.getApi(_) >> apiMock - start() when: def flux = trackAddress.getBalance(Mono.just(req)) then: @@ -101,16 +83,14 @@ class TrackAddressSpec extends Specification { } def blocksBus = TopicProcessor.create() - def upstreamMock = Mock(AggregatedUpstreams) - def headMock = Mock(EthereumHead) - def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackAddress trackAddress = new TrackAddress(upstreams, Schedulers.immediate()) + trackAddress.init() + apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0xff98") - _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock - _ * upstreamMock.getApi(_) >> apiMock - _ * upstreamMock.getHead() >> headMock - _ * headMock.getFlux() >> blocksBus - start() when: def flux = trackAddress.subscribe(Mono.just(req)) then: @@ -120,7 +100,7 @@ class TrackAddressSpec extends Specification { assert trackAddress.isTracked(Chain.ETHEREUM, Address.from(address1)) } .then { - blocksBus.onNext(block2) + upstreamMock.nextBlock(block2) } .expectNext(exp2) .thenCancel() diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy index 1a9dcddf..aecc9908 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy @@ -4,9 +4,7 @@ import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.AggregatedUpstreams -import io.emeraldpay.dshackle.upstream.AvailableChains -import io.emeraldpay.dshackle.upstream.EthereumHead +import io.emeraldpay.dshackle.test.UpstreamsMock import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash @@ -15,7 +13,6 @@ import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionJson import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -25,23 +22,9 @@ import java.time.Instant class TrackTxSpec extends Specification { - AvailableChains availableChains = new AvailableChains(TestingCommons.objectMapper()) - Upstreams upstreams - TrackTx trackTx - def chain = Common.ChainRef.CHAIN_ETHEREUM def txId = "0xba61ce4672751fd6086a9ac2b55547a5555af17535b6c0334ede2ecb6d64070a" - def setup() { - upstreams = Mock(Upstreams) - trackTx = new TrackTx(upstreams, availableChains, Schedulers.immediate()) - } - - def startTrackTxService() { - trackTx.init() - availableChains.add(Chain.ETHEREUM) - availableChains.add(Chain.TESTNET_KOVAN) - } def "Gives details for an old transaction"() { setup: @@ -89,21 +72,15 @@ class TrackTxSpec extends Specification { .setTimestamp(blockJson.timestamp.getTime()) ).build() + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + trackTx.init() - def upstreamMock = Mock(AggregatedUpstreams) - def blocksBus = TopicProcessor.create() - def headMock = Mock(EthereumHead) - - def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) apiMock.answer("eth_getTransactionByHash", [txId], txJson) apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson) - - _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock - _ * upstreamMock.getApi(_) >> apiMock - _ * upstreamMock.getHead() >> headMock - _ * headMock.getFlux() >> blocksBus - _ * headMock.getHead() >> Mono.just(blockHeadJson) - startTrackTxService() + upstreamMock.nextBlock(blockHeadJson) when: def flux = trackTx.add(Mono.just(req)) @@ -127,13 +104,13 @@ class TrackTxSpec extends Specification { .setMined(false) .build() - def upstreamMock = Mock(AggregatedUpstreams) - def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) - apiMock.answer("eth_getTransactionByHash", [txId], null) + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + trackTx.init() - _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock - _ * upstreamMock.getApi(_) >> apiMock - startTrackTxService() + apiMock.answer("eth_getTransactionByHash", [txId], null) when: def act = StepVerifier.withVirtualTime { @@ -184,15 +161,15 @@ class TrackTxSpec extends Specification { it } - def upstreamMock = Mock(AggregatedUpstreams) - def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + trackTx.init() + apiMock.answerOnce("eth_getTransactionByHash", [txId], null) apiMock.answer("eth_getTransactionByHash", [txId], txJson) - _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock - _ * upstreamMock.getApi(_) >> apiMock - startTrackTxService() - when: def act = StepVerifier.withVirtualTime { return trackTx.add(Mono.just(req)) @@ -283,11 +260,12 @@ class TrackTxSpec extends Specification { ) - def upstreamMock = Mock(AggregatedUpstreams) - def blocksBus = TopicProcessor.create() - def headMock = Mock(EthereumHead) + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + trackTx.init() - def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) apiMock.answerOnce("eth_getTransactionByHash", [txId], null) apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted) apiMock.answer("eth_getTransactionByHash", [txId], txJsonMined) @@ -295,20 +273,10 @@ class TrackTxSpec extends Specification { apiMock.answer("eth_getBlockByHash", [block.hash.toHex(), false], block) } - def headBlock = blocks[0] - - _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock - _ * upstreamMock.getApi(_) >> apiMock - _ * upstreamMock.getHead() >> headMock - _ * headMock.getFlux() >> blocksBus - _ * headMock.getHead() >> { return Mono.just(headBlock) } - startTrackTxService() - def nextBlock = { int i -> return { println("block $i"); - headBlock = blocks[i]; - blocksBus.onNext(blocks[i]) + upstreamMock.nextBlock(blocks[i]) } as Runnable } @@ -333,7 +301,11 @@ class TrackTxSpec extends Specification { def "Tracked after first load"() { setup: - startTrackTxService() + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + trackTx.init() def req = BlockchainOuterClass.TxStatusRequest.newBuilder() .setChain(chain) @@ -350,7 +322,12 @@ class TrackTxSpec extends Specification { def "Update of last notified keeps everything else"() { setup: - startTrackTxService() + def apiMock = TestingCommons.api(Stub(RpcClient)) + def upstreamMock = TestingCommons.upstream(apiMock) + Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + trackTx.init() + def req = BlockchainOuterClass.TxStatusRequest.newBuilder() .setChain(chain) .setConfirmationLimit(6) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index 91a0318b..eac1ef7a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -3,10 +3,10 @@ package io.emeraldpay.dshackle.test 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.EthereumApi -import io.emeraldpay.dshackle.upstream.EthereumTargets +import io.emeraldpay.dshackle.upstream.QuorumBasedMethods import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.rpc.RpcClient @@ -23,8 +23,8 @@ class EthereumApiMock extends EthereumApi { List predefined = [] private ObjectMapper objectMapper - EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain, Upstream upstream) { - super(rpcClient, objectMapper, chain, new EthereumTargets(objectMapper, chain), upstream) + EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) { + super(rpcClient, objectMapper, chain, new DirectCallMethods()) this.objectMapper = objectMapper } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy new file mode 100644 index 00000000..166b2344 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -0,0 +1,30 @@ +package io.emeraldpay.dshackle.test + +import io.emeraldpay.dshackle.upstream.EthereumHead +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.TopicProcessor + +class EthereumHeadMock implements EthereumHead { + + private TopicProcessor> bus = TopicProcessor.create() + private BlockJson latest + + void nextBlock(BlockJson block) { + assert block != null + latest = block + bus.onNext(block) + } + + @Override + Mono> getHead() { + return latest != null ? Mono.just(latest) : Mono.from(bus) + } + + @Override + Flux> getFlux() { + return Flux.concat(getHead(), bus).distinctUntilChanged() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy new file mode 100644 index 00000000..7e650cde --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -0,0 +1,35 @@ +package io.emeraldpay.dshackle.test + +import io.emeraldpay.dshackle.upstream.EthereumApi +import io.emeraldpay.dshackle.upstream.EthereumHead +import io.emeraldpay.dshackle.upstream.EthereumUpstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import org.jetbrains.annotations.NotNull + +class EthereumUpstreamMock extends EthereumUpstream { + + EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() + + EthereumUpstreamMock(@NotNull Chain chain, @NotNull EthereumApi api) { + super(chain, api) + setLag(0) + setStatus(UpstreamAvailability.OK) + } + + void nextBlock(BlockJson block) { + ethereumHeadMock.nextBlock(block) + } + + @Override + EthereumHead createHead() { + return ethereumHeadMock + } + + @Override + EthereumHead getHead() { + return ethereumHeadMock + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index c1595530..21aa8ebb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -4,6 +4,8 @@ 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.EthereumApi +import io.emeraldpay.dshackle.upstream.EthereumUpstream import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.Batch @@ -32,11 +34,15 @@ class TestingCommons { return objectMapper } - static EthereumApiMock api(RpcClient rpcClient, Upstream upstream) { - return new EthereumApiMock(rpcClient, objectMapper(), Chain.ETHEREUM, upstream) + static EthereumApiMock api(RpcClient rpcClient) { + return new EthereumApiMock(rpcClient, objectMapper(), Chain.ETHEREUM) } static JacksonRpcConverter rpcConverter() { return new JacksonRpcConverter(objectMapper()) } + + static EthereumUpstreamMock upstream(EthereumApi api) { + return new EthereumUpstreamMock(Chain.ETHEREUM, api) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy new file mode 100644 index 00000000..1ea2bb9d --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy @@ -0,0 +1,65 @@ +package io.emeraldpay.dshackle.test + + +import io.emeraldpay.dshackle.upstream.AggregatedUpstream +import io.emeraldpay.dshackle.upstream.ChainUpstreams +import io.emeraldpay.dshackle.upstream.QuorumBasedMethods +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.grpc.Chain +import org.jetbrains.annotations.NotNull +import reactor.core.publisher.Flux + +class UpstreamsMock implements Upstreams { + + private Map target = [:] + private Map upstreams = [:] + + UpstreamsMock(Chain chain, Upstream up) { + addUpstream(chain, up) + } + UpstreamsMock(Chain chain1, Upstream up1, Chain chain2, Upstream up2) { + addUpstream(chain1, up1) + addUpstream(chain2, up2) + } + + @Override + AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { + if (!upstreams.containsKey(chain)) { + upstreams[chain] = new ChainUpstreams(chain, [up], targetFor(chain)) + } else { + upstreams[chain].addUpstream(up) + } + return upstreams[chain] + } + + @Override + AggregatedUpstream getUpstream(@NotNull Chain chain) { + return upstreams[chain] + } + + @Override + List getAvailable() { + return upstreams.keySet().toList() + } + + @Override + Flux observeChains() { + return Flux.fromIterable(getAvailable()) + } + + @Override + QuorumBasedMethods targetFor(@NotNull Chain chain) { + if (target[chain] == null) { + QuorumBasedMethods targets = new QuorumBasedMethods(TestingCommons.objectMapper(), chain) + target[chain] = targets + } + return target[chain] + } + + @Override + boolean isAvailable(@NotNull Chain chain) { + return upstreams.containsKey(chain) + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index 9d0c8a7d..037f2306 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -19,15 +19,16 @@ class EthereumGrpcTransportSpec extends Specification { MockServer mockServer = new MockServer() ObjectMapper objectMapper = TestingCommons.objectMapper() - def ethereumTargets = new EthereumTargets(objectMapper, Chain.ETHEREUM) + def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) def "Make simple call"() { setup: def callData = [:] def otherSideUpstreams = Mock(Upstreams) - def otherSideAggr = Mock(AggregatedUpstreams) + def otherSideAggr = Mock(AggregatedUpstream) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) - def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM, otherSideAggr) + def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) + otherSideApi.upstream = otherSideAggr def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @Override @@ -68,9 +69,10 @@ class EthereumGrpcTransportSpec extends Specification { setup: def callData = [:] def otherSideUpstreams = Mock(Upstreams) - def otherSideAggr = Mock(AggregatedUpstreams) + def otherSideAggr = Mock(AggregatedUpstream) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) - def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM, otherSideAggr) + def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) + otherSideApi.upstream = otherSideAggr def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy index f1c07143..a2a519f4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy @@ -10,7 +10,7 @@ class FilteringApiIteratorSpec extends Specification { def rpcClient = new DefaultRpcClient(null) def objectMapper = TestingCommons.objectMapper() - def ethereumTargets = new EthereumTargets(objectMapper, Chain.ETHEREUM) + def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) def "Verifies labels"() { setup: @@ -23,7 +23,7 @@ class FilteringApiIteratorSpec extends Specification { ].collect { new EthereumUpstream( Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM, ethereumTargets, null), + new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM, ethereumTargets), (EthereumWs) null, new UpstreamsConfig.Options(), new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)), diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy index cd3dfde6..7050c7b4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy @@ -5,7 +5,6 @@ import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.test.EthereumApiMock import io.emeraldpay.dshackle.test.MockServer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.grpc.Chain @@ -13,7 +12,6 @@ import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.json.BlockJson -import org.apache.commons.codec.binary.Hex import spock.lang.Specification import java.time.Duration @@ -23,13 +21,13 @@ class GrpcUpstreamSpec extends Specification { MockServer mockServer = new MockServer() ObjectMapper objectMapper = TestingCommons.objectMapper() - def ethereumTargets = new EthereumTargets(objectMapper, Chain.ETHEREUM) + def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) def "Subscribe to head"() { setup: def callData = [:] def chain = Chain.ETHEREUM - def api = TestingCommons.api(Stub(RpcClient), Stub(Upstream)) + def api = TestingCommons.api(Stub(RpcClient)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -71,7 +69,7 @@ class GrpcUpstreamSpec extends Specification { def callData = [:] def finished = new CompletableFuture() def chain = Chain.ETHEREUM - def api = TestingCommons.api(Stub(RpcClient), Stub(Upstream)) + def api = TestingCommons.api(Stub(RpcClient)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -128,7 +126,7 @@ class GrpcUpstreamSpec extends Specification { def callData = [:] def finished = new CompletableFuture() def chain = Chain.ETHEREUM - def api = TestingCommons.api(Stub(RpcClient), Stub(Upstream)) + def api = TestingCommons.api(Stub(RpcClient)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")