problem: doesn't update head subscription through remote dshackle, because upstream becomes ready later after chain setup

solution: fix and refactor to avoid such situation in code

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<CallContext<Tuple2<String, String>>> {
+    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<Tuple2<String, String>>> {
         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)
                 }

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 upstreams: Upstreams
-        @Autowired private val availableChains: AvailableChains
 ) {
 
     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 {

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 upstreams: Upstreams
-        @Autowired private val availableChains: AvailableChains
 ) {
 
     fun subscribeStatus(requestMono: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
         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)
 
 }

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<SimpleAddress> {
         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") {

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()

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(
+abstract class AggregatedUpstream(
-        val targets: EthereumTargets
+        val targets: CallMethods
 ): Upstream {
 
     abstract fun getAll(): List<Upstream>

src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt

@@ -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<Chain>()
-    private val bus = TopicProcessor.create<Chain>()
-    private val callTargets = HashMap<Chain, EthereumTargets>()
-
-    fun add(chain: Chain) {
-        if (all.contains(chain)) {
-            return
-        }
-        all.add(chain)
-        bus.onNext(chain)
-    }
-
-    fun observe(): Flux<Chain> {
-        return Flux.merge(
-                Flux.fromIterable(all),
-                Flux.from(bus)
-        )
-    }
-
-    fun supports(chain: Chain): Boolean {
-        return all.contains(chain)
-    }
-
-    fun getAll(): Set<Chain> {
-        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
-    }
-}

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<String>
+    fun isHardcoded(method: String): Boolean
+    fun hardcoded(method: String): Any
+}

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<Upstream>,
-        targets: EthereumTargets
+        targets: CallMethods
-) : AggregatedUpstreams(targets) {
+) : AggregatedUpstream(targets) {
 
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0

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 objectMapper: ObjectMapper
-        @Autowired private val availableChains: AvailableChains
 ) : Upstreams {
 
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
     private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams>()
+    private val chainsBus = TopicProcessor.create<Chain>()
+    private val callTargets = HashMap<Chain, QuorumBasedMethods>()
 
     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(chain, EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain)))
-                    .addUpstream(
-                            EthereumUpstream(
-                                    chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), availableChains.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<Upstream>(), targetFor(chain))
-            val created = ChainUpstreams(chain, ArrayList<Upstream>(), availableChains.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<Chain> {
         return Collections.unmodifiableList(chainMapping.keys.toList())
     }
+
+    override fun observeChains(): Flux<Chain> {
+        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)
+    }
 }

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,
+        defaultLag: Long,
-        avail: UpstreamAvailability
+        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<UpstreamAvailability> = TopicProcessor.create()
 
     override fun getStatus(): UpstreamAvailability {

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<String> {
+        return emptySet()
+    }
+
+    override fun isHardcoded(method: String): Boolean {
+        return false
+    }
+
+    override fun hardcoded(method: String): Any {
+        return "unsupported"
+    }
+}

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,
+        val targets: CallMethods
-        var upstream: Upstream? = null
 ) {
 
     private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
+    var upstream: Upstream? = null
 
     private val timeout = Duration.ofSeconds(5)
     private val log = LoggerFactory.getLogger(EthereumApi::class.java)

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<String> {
         return targets.getSupportedMethods()
     }
 
     private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
 
-    private val head: EthereumHead = if (ethereumWs != null) {
+    private val head: EthereumHead = createHead()
-        EthereumWsHead(ethereumWs)
-    } else {
-        EthereumRpcHead(api).apply {
-            this.start()
-        }
-    }
 
     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)
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt

@@ -1,30 +1,30 @@
 package io.emeraldpay.dshackle.upstream
 
 class FilteringApiIterator(
-        private val apis: List<Upstream>,
+        private val upstreams: List<Upstream>,
         private var pos: Int,
         private val matcher: Selector.Matcher,
         private val repeatLimit: Int = 3
 ): Iterator<EthereumApi> {
 
-    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]
+            val upstream = upstreams[pos++ % upstreams.size]
-            if (api.isAvailable(matcher)) {
+            if (upstream.isAvailable(matcher)) {
-                nextApi = api
+                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!!
+            val curr = nextUpstream!!
-            nextApi = null
+            nextUpstream = null
             return curr.getApi(matcher)
         }
         throw IllegalStateException("No upstream API available")

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() {

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 {

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<String> {
+    override fun getSupportedMethods(): Set<String> {
         return allowedMethods.plus(hardcodedMethods).toSortedSet()
     }
 }

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 addUpstream(chain: Chain, up: Upstream): AggregatedUpstream
-    fun getUpstream(chain: Chain): AggregatedUpstreams?
+    fun getUpstream(chain: Chain): AggregatedUpstream?
     fun getAvailable(): List<Chain>
+    fun observeChains(): Flux<Chain>
+    fun targetFor(chain: Chain): CallMethods
+    fun isAvailable(chain: Chain): Boolean
 }

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)
-        def apiMock = TestingCommons.api(rpcClient, upstream)
+        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)
-        def apiMock = TestingCommons.api(rpcClient, upstream)
+        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)
-        def apiMock = TestingCommons.api(rpcClient, upstream)
+        apiMock.upstream = Stub(Upstream)
 
         apiMock.answer("eth_test", [], null, 3)
         apiMock.answerOnce("eth_test", [], "foo")

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.test.UpstreamsMock
-import io.emeraldpay.dshackle.upstream.AvailableChains
-import io.emeraldpay.dshackle.upstream.EthereumHead
 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))
-        def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock)
+        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 apiMock = TestingCommons.api(Stub(RpcClient))
-        def headMock = Mock(EthereumHead)
+        def upstreamMock = TestingCommons.upstream(apiMock)
-        def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock)
+        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()

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.test.UpstreamsMock
-import io.emeraldpay.dshackle.upstream.AvailableChains
-import io.emeraldpay.dshackle.upstream.EthereumHead
 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)
-
+        upstreamMock.nextBlock(blockHeadJson)
-        _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock
-        _ * upstreamMock.getApi(_) >> apiMock
-        _ * upstreamMock.getHead() >> headMock
-        _ * headMock.getFlux() >> blocksBus
-        _ * headMock.getHead() >> Mono.just(blockHeadJson)
-        startTrackTxService()
 
         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))
-        def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock)
+        def upstreamMock = TestingCommons.upstream(apiMock)
-        apiMock.answer("eth_getTransactionByHash", [txId], null)
+        Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
+        TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate())
+        trackTx.init()
 
-        _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock
+        apiMock.answer("eth_getTransactionByHash", [txId], null)
-        _ * upstreamMock.getApi(_) >> apiMock
-        startTrackTxService()
 
         when:
         def act = StepVerifier.withVirtualTime {
@@ -184,15 +161,15 @@ class TrackTxSpec extends Specification {
             it
         }
 
-        def upstreamMock = Mock(AggregatedUpstreams)
+        def apiMock = TestingCommons.api(Stub(RpcClient))
-        def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock)
+        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 apiMock = TestingCommons.api(Stub(RpcClient))
-        def blocksBus = TopicProcessor.create()
+        def upstreamMock = TestingCommons.upstream(apiMock)
-        def headMock = Mock(EthereumHead)
+        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];
+                upstreamMock.nextBlock(blocks[i])
-                blocksBus.onNext(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)

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<PredefinedResponse> predefined = []
     private ObjectMapper objectMapper
 
-    EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain, Upstream upstream) {
+    EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) {
-        super(rpcClient, objectMapper, chain, new EthereumTargets(objectMapper, chain), upstream)
+        super(rpcClient, objectMapper, chain, new DirectCallMethods())
         this.objectMapper = objectMapper
     }
 

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<BlockJson<TransactionId>> bus = TopicProcessor.create()
+    private BlockJson<TransactionId> latest
+
+    void nextBlock(BlockJson<TransactionId> block) {
+        assert block != null
+        latest = block
+        bus.onNext(block)
+    }
+
+    @Override
+    Mono<BlockJson<TransactionId>> getHead() {
+        return latest != null ? Mono.just(latest) : Mono.from(bus)
+    }
+
+    @Override
+    Flux<BlockJson<TransactionId>> getFlux() {
+        return Flux.concat(getHead(), bus).distinctUntilChanged()
+    }
+}

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<TransactionId> block) {
+        ethereumHeadMock.nextBlock(block)
+    }
+
+    @Override
+    EthereumHead createHead() {
+        return ethereumHeadMock
+    }
+
+    @Override
+    EthereumHead getHead() {
+        return ethereumHeadMock
+    }
+}

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) {
+    static EthereumApiMock api(RpcClient rpcClient) {
-        return new EthereumApiMock(rpcClient, objectMapper(), Chain.ETHEREUM, upstream)
+        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)
+    }
 }

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<Chain, QuorumBasedMethods> target = [:]
+    private Map<Chain, AggregatedUpstream> 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<Chain> getAvailable() {
+        return upstreams.keySet().toList()
+    }
+
+    @Override
+    Flux<Chain> 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)
+    }
+
+}

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

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)),

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<Boolean>()
         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<Boolean>()
         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")