solution: select which kind of upstream to use for native call

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -94,6 +94,16 @@ class UpstreamsConfig {
 
     //TODO make it unmodifiable after initial load
     class Labels: HashMap<String, String>() {
+
+        companion object {
+            @JvmStatic fun fromMap(map: Map<String, String>): Labels {
+                val labels = Labels()
+                map.entries.forEach() { kv ->
+                    labels.put(kv.key, kv.value)
+                }
+                return labels
+            }
+        }
     }
 
     enum class UpstreamType private constructor(vararg code: String) {

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -5,9 +5,11 @@ import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
 import io.emeraldpay.dshackle.upstream.EthereumApi
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
+import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
@@ -30,10 +32,11 @@ class NativeCall(
         return requestMono.flatMapMany { request ->
             val chain= Chain.byId(request.chain.number)
             if (chain == Chain.UNSPECIFIED) {
+                // TODO send error to all requests?
                 throw Exception("Invalid chain id: ${request.chain.number}")
             }
-            // TODO send error to all requests?
+            val matcher = Selector.convertToMatcher(request.selector)
-            val upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable")
+            val upstream = upstreams.getUpstream(chain)?.getApi(matcher) ?: throw Exception("Chain ${chain.id} is unavailable")
             request.itemsList.toFlux().map {
                 val method = it.target
                 val params = it.payload.toStringUtf8()

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt

@@ -3,6 +3,7 @@ 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
 import io.infinitape.etherjar.domain.Address
@@ -68,7 +69,9 @@ class TrackAddress(
     }
 
     private fun stopTracking(client: TrackedAddress) {
-        clients[client.chain]?.remove(client) ?: log.warn("Chain ${client.chain} is not available for tracking")
+        clients[client.chain]?.removeIf {
+            it.id == client.id
+        } ?: log.warn("Chain ${client.chain} is not available for tracking")
     }
 
     fun isTracked(chain: Chain, address: Address): Boolean {
@@ -151,7 +154,7 @@ class TrackAddress(
 
     fun getBalance(addr: SimpleAddress): Mono<Wei> {
         val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}"))
-        return up.getApi()
+        return up.getApi(Selector.empty)
                 .executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST))
                 .timeout(Duration.ofSeconds(15))
     }
@@ -208,9 +211,5 @@ class TrackAddress(
                          val id: Long
     ): SimpleAddress(chain, address, balance) {
         override fun withBalance(balance: Wei) = TrackedAddress(chain, stream, address, lastPing, balance, id)
-
-        override fun equals(other: Any?): Boolean {
-            return other != null && other is TrackedAddress && other.id == id
-        }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt

@@ -4,6 +4,7 @@ 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.Upstreams
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.BlockHash
@@ -99,7 +100,7 @@ class TrackTx(
     private fun loadWeight(tx: TrackedTx): Mono<TrackedTx> {
         val upstream = upstreams.getUpstream(tx.chain)
                 ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}"))
-        return upstream.getApi()
+        return upstream.getApi(Selector.empty)
                 .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash))
                 .map { block ->
                     if (block != null && block.number != null && block.totalDifficulty != null) {
@@ -119,7 +120,7 @@ class TrackTx(
 
     private fun checkForUpdate(tx: TrackedTx): Mono<TrackedTx> {
         val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}"))
-        val execution = upstream.getApi()
+        val execution = upstream.getApi(Selector.empty)
                 .executeAndConvert(Commands.eth().getTransaction(tx.txid))
         return execution.flatMap {
             if (it.blockNumber != null && it.blockHash != null && it.blockHash != ZERO_BLOCK) {

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

@@ -11,7 +11,7 @@ abstract class AggregatedUpstreams: Upstream {
 
     abstract fun getAll(): List<Upstream>
     abstract fun addUpstream(upstream: Upstream)
-    abstract fun getApis(quorum: Int): Iterator<EthereumApi>
+    abstract fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator<EthereumApi>
 
     override fun observeStatus(): Flux<UpstreamAvailability> {
         val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } }
@@ -28,8 +28,8 @@ abstract class AggregatedUpstreams: Upstream {
         return list
     }
 
-    override fun isAvailable(): Boolean {
+    override fun isAvailable(matcher: Selector.Matcher): Boolean {
-        return getAll().any { it.isAvailable() }
+        return getAll().any { it.isAvailable(matcher) }
     }
 
     override fun getStatus(): UpstreamAvailability {
@@ -39,54 +39,7 @@ abstract class AggregatedUpstreams: Upstream {
     }
 
     override fun getOptions(): UpstreamsConfig.Options {
-        val options = UpstreamsConfig.Options()
+        return UpstreamsConfig.Options()
-        options.quorum = getAll().filter {
-            it.getStatus() == UpstreamAvailability.OK
-        }.sumBy {
-            it.getOptions().quorum
-        }
-        return options
-    }
-
-    class SingleApi(
-            private val quorumApi: QuorumApi
-    ): Iterator<EthereumApi> {
-
-        private var consumed = false
-
-        override fun hasNext(): Boolean {
-            return !consumed && quorumApi.hasNext()
-        }
-
-        override fun next(): EthereumApi {
-            consumed = true
-            return quorumApi.next()
-        }
-    }
-
-    class QuorumApi(
-            private val apis: List<Upstream>,
-            private val quorum: Int,
-            private var pos: Int
-    ): Iterator<EthereumApi> {
-
-        private var consumed = 0
-
-        override fun hasNext(): Boolean {
-            return consumed < quorum
-        }
-
-        override fun next(): EthereumApi {
-            val start = pos
-            while (pos < start + apis.size) {
-                val api = apis[pos++ % apis.size]
-                if (api.isAvailable()) {
-                    consumed++
-                    return api.getApi()
-                }
-            }
-            throw IllegalStateException("No upstream API available")
-        }
     }
 
     class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())

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

@@ -1,6 +1,5 @@
 package io.emeraldpay.dshackle.upstream
 
-import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import java.io.Closeable
@@ -44,16 +43,16 @@ class ChainUpstreams (
         head = updateHead()
     }
 
-    override fun getApis(quorum: Int): Iterator<EthereumApi> {
+    override fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator<EthereumApi> {
         val i = seq++
         if (seq >= Int.MAX_VALUE / 2) {
             seq = 0
         }
-        return QuorumApi(upstreams, 1, seq)
+        return FilteringApiIterator(upstreams, 1, seq, matcher)
     }
 
-    override fun getApi(): EthereumApi {
+    override fun getApi(matcher: Selector.Matcher): EthereumApi {
-        return getApis(1).next()
+        return getApis(1, matcher).next()
     }
 
     override fun getHead(): EthereumHead {

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

@@ -12,7 +12,6 @@ import reactor.core.publisher.Mono
 import reactor.core.publisher.toFlux
 import java.time.Duration
 import java.util.*
-import java.util.concurrent.CompletableFuture
 
 open class EthereumApi(
         val rpcClient: RpcClient,

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

@@ -21,14 +21,27 @@ import java.util.concurrent.CompletableFuture
 import java.util.function.Function
 
 class EthereumGrpcTransport(
-        private val chain: Chain,
+        private val chainRef: Common.ChainRef,
+        private val selector: BlockchainOuterClass.Selector?,
         private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val objectMapper: ObjectMapper
 ): RpcTransport {
 
-    private val chainRef = Common.ChainRef.forNumber(chain.id)
     private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
 
+    constructor(
+            chain: Chain,
+            client: ReactorBlockchainGrpc.ReactorBlockchainStub,
+            objectMapper: ObjectMapper
+    ) : this(Common.ChainRef.forNumber(chain.id), Selector.EmptyMatcher().asProto(), client, objectMapper)
+
+    fun withMatcher(matcher: Selector.Matcher): EthereumGrpcTransport {
+        if (matcher is Selector.EmptyMatcher && selector == null) {
+            return this
+        }
+        return EthereumGrpcTransport(chainRef, matcher.asProto(), client, objectMapper)
+    }
+
     override fun close() {
     }
 
@@ -87,7 +100,10 @@ class EthereumGrpcTransport(
 
     override fun execute(items: List<Batch.BatchItem<out Any, out Any>>): CompletableFuture<BatchStatus> {
         val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
-                .setChain(chainRef);
+                .setChain(chainRef)
+        if (selector != null) {
+            req.setSelector(selector)
+        }
         val mapping = prepareMapping(items, req)
         return client.nativeCall(req.build())
                 .map(replyProcessor(mapping))

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

@@ -9,7 +9,7 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.TopicProcessor
 import java.util.concurrent.atomic.AtomicReference
 
-class EthereumUpstream(
+open class EthereumUpstream(
         val chain: Chain,
         private val api: EthereumApi,
         private val ethereumWs: EthereumWs? = null,
@@ -45,14 +45,18 @@ class EthereumUpstream(
                 }
     }
 
-    override fun isAvailable(): Boolean {
+    override fun isAvailable(matcher: Selector.Matcher): Boolean {
-        return status.get() == UpstreamAvailability.OK
+        return status.get() == UpstreamAvailability.OK && matcher.matches(node.labels)
     }
 
     override fun getStatus(): UpstreamAvailability {
         return status.get()
     }
 
+    fun setStatus(avail: UpstreamAvailability) {
+        status.set(avail)
+    }
+
     override fun observeStatus(): Flux<UpstreamAvailability> {
         return Flux.from(statusStream)
     }
@@ -61,7 +65,11 @@ class EthereumUpstream(
         return head
     }
 
-    override fun getApi(): EthereumApi {
+    override fun getApi(matcher: Selector.Matcher): EthereumApi {
+        return api
+    }
+
+    fun getApi(): EthereumApi {
         return api
     }
 

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

@@ -0,0 +1,27 @@
+package io.emeraldpay.dshackle.upstream
+
+class FilteringApiIterator(
+        private val apis: List<Upstream>,
+        private val quorum: Int,
+        private var pos: Int,
+        private val matcher: Selector.Matcher
+): Iterator<EthereumApi> {
+
+    private var consumed = 0
+
+    override fun hasNext(): Boolean {
+        return consumed < quorum
+    }
+
+    override fun next(): EthereumApi {
+        val start = pos
+        while (pos < start + apis.size) {
+            val api = apis[pos++ % apis.size]
+            if (api.isAvailable(matcher)) {
+                consumed++
+                return api.getApi(matcher)
+            }
+        }
+        throw IllegalStateException("No upstream API available")
+    }
+}

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

@@ -41,14 +41,13 @@ open class GrpcUpstream(
     private val status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)
     private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
     private val head = Head(this)
-    private val api: EthereumApi
     private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
     private val supportedMethods = HashSet<String>()
+    private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
 
-    init {
+    open fun createApi(matcher: Selector.Matcher): EthereumApi {
-        val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
+        val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher))
-        val rpcClient = DefaultRpcClient(grpcTransport)
+        return EthereumApi(rpcClient, objectMapper, chain)
-        api = EthereumApi(rpcClient, objectMapper, chain)
     }
 
     open fun connect() {
@@ -80,7 +79,7 @@ open class GrpcUpstream(
                     curr == null || curr.totalDifficulty < block.totalDifficulty
                 }
                 .flatMap {
-                    getApi()
+                    getApi(Selector.EmptyMatcher())
                             .executeAndConvert(Commands.eth().getBlock(it.hash))
                             .timeout(Duration.ofSeconds(15))
                 }
@@ -135,8 +134,10 @@ open class GrpcUpstream(
         return supportedMethods
     }
 
-    override fun isAvailable(): Boolean {
+    override fun isAvailable(matcher: Selector.Matcher): Boolean {
-        return headBlock.get() != null
+        return headBlock.get() != null && nodes.get().getNodes().any {
+            it.quorum > 0 && matcher.matches(it.labels)
+        }
     }
 
     override fun getStatus(): UpstreamAvailability {
@@ -151,8 +152,8 @@ open class GrpcUpstream(
         return head
     }
 
-    override fun getApi(): EthereumApi {
+    override fun getApi(matcher: Selector.Matcher): EthereumApi {
-        return api
+        return createApi(matcher)
     }
 
     override fun getOptions(): UpstreamsConfig.Options {

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

@@ -0,0 +1,128 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import org.apache.commons.lang3.StringUtils
+import java.util.*
+
+class Selector {
+
+    companion object {
+
+        val empty = EmptyMatcher()
+
+        @JvmStatic
+        fun convertToMatcher(req: BlockchainOuterClass.Selector?): Matcher {
+            return when {
+                req == null -> EmptyMatcher()
+                req.hasLabelSelector() -> req.labelSelector.let { selector ->
+                    if (StringUtils.isNotEmpty(selector.name)) {
+                        val values = selector.valueList
+                                .map { it?.trim() ?: "" }
+                                .filter { StringUtils.isNotEmpty(it) }
+                        if (values.isEmpty()) {
+                            ExistsMatcher(selector.name)
+                        } else {
+                            LabelMatcher(selector.name, selector.valueList)
+                        }
+                    } else {
+                        EmptyMatcher()
+                    }
+                }
+                req.hasAndSelector() -> AndMatcher(Collections.unmodifiableCollection(req.andSelector.selectorsList.map { convertToMatcher(it) }))
+                req.hasOrSelector() -> OrMatcher(Collections.unmodifiableCollection(req.orSelector.selectorsList.map { convertToMatcher(it) }))
+                req.hasNotSelector() -> NotMatcher(convertToMatcher(req.notSelector.selector))
+                req.hasExistsSelector() -> ExistsMatcher(req.existsSelector.name)
+                else -> EmptyMatcher()
+            }
+        }
+    }
+
+    interface Matcher {
+        fun matches(labels: UpstreamsConfig.Labels): Boolean
+        fun asProto(): BlockchainOuterClass.Selector?
+    }
+
+    class EmptyMatcher: Matcher {
+        override fun matches(labels: UpstreamsConfig.Labels): Boolean {
+            return true
+        }
+
+        override fun asProto(): BlockchainOuterClass.Selector? {
+            return null
+        }
+    }
+
+    class LabelMatcher(val name: String, val values: Collection<String>): Matcher {
+        override fun matches(labels: UpstreamsConfig.Labels): Boolean {
+            return labels.get(name)?.let {
+                labelValue -> values.any { it == labelValue }
+            } ?: false
+        }
+
+        override fun asProto(): BlockchainOuterClass.Selector {
+            return BlockchainOuterClass.Selector.newBuilder().setLabelSelector(
+                    BlockchainOuterClass.LabelSelector.newBuilder()
+                            .setName(name)
+                            .addAllValue(values)
+            ).build()
+        }
+    }
+
+    class OrMatcher(val matchers: Collection<Matcher>): Matcher {
+        override fun matches(labels: UpstreamsConfig.Labels): Boolean {
+            return matchers.any { matcher -> matcher.matches(labels) }
+        }
+
+        override fun asProto(): BlockchainOuterClass.Selector {
+            return BlockchainOuterClass.Selector.newBuilder().setOrSelector(
+                    BlockchainOuterClass.OrSelector.newBuilder()
+                            .addAllSelectors(matchers.map { it.asProto() })
+                            .build()
+            ).build()
+        }
+    }
+
+    class AndMatcher(val matchers: Collection<Matcher>): Matcher {
+        override fun matches(labels: UpstreamsConfig.Labels): Boolean {
+            return matchers.all { matcher -> matcher.matches(labels) }
+        }
+
+        override fun asProto(): BlockchainOuterClass.Selector {
+            return BlockchainOuterClass.Selector.newBuilder().setAndSelector(
+                    BlockchainOuterClass.AndSelector.newBuilder()
+                            .addAllSelectors(matchers.map { it.asProto() })
+                            .build()
+            ).build()
+        }
+    }
+
+    class NotMatcher(val matcher: Matcher): Matcher {
+        override fun matches(labels: UpstreamsConfig.Labels): Boolean {
+            return !matcher.matches(labels)
+        }
+
+        override fun asProto(): BlockchainOuterClass.Selector {
+            return BlockchainOuterClass.Selector.newBuilder().setNotSelector(
+                    BlockchainOuterClass.NotSelector.newBuilder()
+                            .setSelector(matcher.asProto())
+                            .build()
+            ).build()
+        }
+    }
+
+    class ExistsMatcher(val name: String): Matcher {
+        override fun matches(labels: UpstreamsConfig.Labels): Boolean {
+            return labels.containsKey(name)
+        }
+
+        override fun asProto(): BlockchainOuterClass.Selector {
+            return BlockchainOuterClass.Selector.newBuilder().setExistsSelector(
+                    BlockchainOuterClass.ExistsSelector.newBuilder()
+                            .setName(name)
+                            .build()
+            ).build()
+        }
+    }
+
+}

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

@@ -4,11 +4,11 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import reactor.core.publisher.Flux
 
 interface Upstream {
-    fun isAvailable(): Boolean
+    fun isAvailable(matcher: Selector.Matcher): Boolean
     fun getStatus(): UpstreamAvailability
     fun observeStatus(): Flux<UpstreamAvailability>
     fun getHead(): EthereumHead
-    fun getApi(): EthereumApi
+    fun getApi(matcher: Selector.Matcher): EthereumApi
     fun getOptions(): UpstreamsConfig.Options
     fun getSupportedTargets(): Set<String>
 }

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

@@ -17,7 +17,7 @@ class UpstreamValidator(
         val peerCount = batch.add(Commands.net().peerCount())
         val syncing = batch.add(Commands.eth().syncing())
         try {
-            ethereumUpstream.getApi().rpcClient.execute(batch).get(5, TimeUnit.SECONDS)
+            ethereumUpstream.getApi(Selector.empty).rpcClient.execute(batch).get(5, TimeUnit.SECONDS)
             if (syncing.get().isSyncing) {
                 return UpstreamAvailability.SYNCING
             }

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy

@@ -23,7 +23,7 @@ import java.time.Duration
 
 class TrackAddressSpec extends Specification {
 
-    AvailableChains availableChains = new AvailableChains()
+    AvailableChains availableChains
     Upstreams upstreams
     TrackAddress trackAddress
 
@@ -37,6 +37,7 @@ class TrackAddressSpec extends Specification {
 
 
     def setup() {
+        availableChains = new AvailableChains()
         upstreams = Mock(Upstreams)
         trackAddress = new TrackAddress(upstreams, availableChains, Schedulers.immediate())
     }
@@ -63,7 +64,7 @@ class TrackAddressSpec extends Specification {
         def apiMock = new EthereumApiMock(Mock(RpcClient), TestingCommons.objectMapper(), Chain.ETHEREUM)
         apiMock.answer("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2")
         _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock
-        _ * upstreamMock.getApi() >> apiMock
+        _ * upstreamMock.getApi(_) >> apiMock
         start()
         when:
         def flux = trackAddress.getBalance(Mono.just(req))
@@ -106,7 +107,7 @@ class TrackAddressSpec extends Specification {
         apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2")
         apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0xff98")
         _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock
-        _ * upstreamMock.getApi() >> apiMock
+        _ * upstreamMock.getApi(_) >> apiMock
         _ * upstreamMock.getHead() >> headMock
         _ * headMock.getFlux() >> blocksBus
         start()
@@ -124,6 +125,7 @@ class TrackAddressSpec extends Specification {
                 .expectNext(exp2)
                 .thenCancel()
                 .verify(Duration.ofSeconds(3))
+        Thread.sleep(50)
         !trackAddress.isTracked(Chain.ETHEREUM, Address.from(address1))
     }
 }

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy

@@ -99,7 +99,7 @@ class TrackTxSpec extends Specification {
         apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson)
 
         _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock
-        _ * upstreamMock.getApi() >> apiMock
+        _ * upstreamMock.getApi(_) >> apiMock
         _ * upstreamMock.getHead() >> headMock
         _ * headMock.getFlux() >> blocksBus
         _ * headMock.getHead() >> Mono.just(blockHeadJson)
@@ -182,7 +182,7 @@ class TrackTxSpec extends Specification {
         def headBlock = blocks[0]
 
         _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock
-        _ * upstreamMock.getApi() >> apiMock
+        _ * upstreamMock.getApi(_) >> apiMock
         _ * upstreamMock.getHead() >> headMock
         _ * headMock.getFlux() >> blocksBus
         _ * headMock.getHead() >> { return Mono.just(headBlock) }

src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy

@@ -45,7 +45,7 @@ class EthereumGrpcTransportSpec extends Specification {
 
         then:
         1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr
-        1 * otherSideAggr.api >> otherSideApi
+        1 * otherSideAggr.getApi(_) >> otherSideApi
         status.failed == 0
         status.succeed == 1
         status.total == 1
@@ -90,7 +90,7 @@ class EthereumGrpcTransportSpec extends Specification {
 
         then:
         1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr
-        1 * otherSideAggr.api >> otherSideApi
+        1 * otherSideAggr.getApi(_) >> otherSideApi
         status.failed == 0
         status.succeed == 2
         status.total == 2

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy

@@ -0,0 +1,88 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.rpc.DefaultRpcClient
+import spock.lang.Specification
+
+class FilteringApiIteratorSpec extends Specification {
+
+    def rpcClient = new DefaultRpcClient(null)
+    def objectMapper = TestingCommons.objectMapper()
+
+    def "Verifies labels"() {
+        setup:
+        def upstreams = [
+               new EthereumUpstream(
+                       Chain.ETHEREUM,
+                       new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM),
+                       (EthereumWs)null,
+                       new UpstreamsConfig.Options(),
+                       new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "foo"]))
+               ),
+               new EthereumUpstream(
+                       Chain.ETHEREUM,
+                       new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM),
+                       (EthereumWs)null,
+                       new UpstreamsConfig.Options(),
+                       new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "bar"]))
+               ),
+               new EthereumUpstream(
+                       Chain.ETHEREUM,
+                       new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM),
+                       (EthereumWs)null,
+                       new UpstreamsConfig.Options(),
+                       new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "foo", test2: "baz"]))
+               ),
+               new EthereumUpstream(
+                       Chain.ETHEREUM,
+                       new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM),
+                       (EthereumWs)null,
+                       new UpstreamsConfig.Options(),
+                       new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "foo"]))
+               ),
+               new EthereumUpstream(
+                       Chain.ETHEREUM,
+                       new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM),
+                       (EthereumWs)null,
+                       new UpstreamsConfig.Options(),
+                       new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "baz"]))
+               )
+        ]
+        def matcher = new Selector.LabelMatcher("test", ["foo"])
+        upstreams.forEach {
+            it.setStatus(UpstreamAvailability.OK)
+        }
+        when:
+        def iter = new FilteringApiIterator(upstreams, 3, 0, matcher)
+        then:
+        iter.hasNext()
+        iter.next() == upstreams[0].api
+        iter.hasNext()
+        iter.next() == upstreams[2].api
+        iter.hasNext()
+        iter.next() == upstreams[3].api
+        !iter.hasNext()
+
+        when:
+        iter = new FilteringApiIterator(upstreams, 2, 1, matcher)
+        then:
+        iter.hasNext()
+        iter.next() == upstreams[2].api
+        iter.hasNext()
+        iter.next() == upstreams[3].api
+        !iter.hasNext()
+
+        when:
+        iter = new FilteringApiIterator(upstreams, 3, 1, matcher)
+        then:
+        iter.hasNext()
+        iter.next() == upstreams[2].api
+        iter.hasNext()
+        iter.next() == upstreams[3].api
+        iter.hasNext()
+        iter.next() == upstreams[0].api
+        !iter.hasNext()
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy

@@ -0,0 +1,279 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import spock.lang.Specification
+
+class SelectorSpec extends Specification {
+
+    private BlockchainOuterClass.LabelSelector.Builder selectLabel1 = BlockchainOuterClass.LabelSelector.newBuilder()
+            .setName("foo").addAllValue(["bar"])
+    private BlockchainOuterClass.Selector selectLabel1Selector = BlockchainOuterClass.Selector.newBuilder()
+            .setLabelSelector(selectLabel1).build()
+
+    private BlockchainOuterClass.LabelSelector.Builder selectLabel2 = BlockchainOuterClass.LabelSelector.newBuilder()
+            .setName("baz").addAllValue(["bar"])
+    private BlockchainOuterClass.Selector selectLabel2Selector = BlockchainOuterClass.Selector.newBuilder()
+            .setLabelSelector(selectLabel2).build()
+
+    def "Convert nothing"() {
+        when:
+        def act = Selector.convertToMatcher(null)
+        then:
+        act.class == Selector.EmptyMatcher
+    }
+
+    def "Convert LABEL match"() {
+        when:
+        def act = Selector.convertToMatcher(
+                selectLabel1Selector
+        )
+        then:
+        act instanceof Selector.LabelMatcher
+        with((Selector.LabelMatcher)act) {
+            name == "foo"
+            values == ["bar"]
+        }
+    }
+
+    def "Convert EXISTS match"() {
+        when:
+        def act = Selector.convertToMatcher(
+                BlockchainOuterClass.Selector.newBuilder()
+                        .setExistsSelector(
+                                BlockchainOuterClass.ExistsSelector.newBuilder()
+                                        .setName("foo")
+                        ).build()
+        )
+        then:
+        act instanceof Selector.ExistsMatcher
+        with((Selector.ExistsMatcher)act) {
+            name == "foo"
+        }
+    }
+
+    def "Convert NOT match"() {
+        when:
+        def act = Selector.convertToMatcher(
+                BlockchainOuterClass.Selector.newBuilder()
+                        .setNotSelector(
+                                BlockchainOuterClass.NotSelector.newBuilder()
+                                        .setSelector(selectLabel1Selector).build()
+                        )
+                        .build()
+        )
+        then:
+        act instanceof Selector.NotMatcher
+        with((Selector.NotMatcher)act) {
+            matcher instanceof Selector.LabelMatcher
+            with((Selector.LabelMatcher)matcher) {
+                name == "foo"
+                values == ["bar"]
+            }
+        }
+    }
+
+    def "Convert OR match"() {
+        when:
+        def act = Selector.convertToMatcher(
+                BlockchainOuterClass.Selector.newBuilder()
+                        .setOrSelector(
+                                BlockchainOuterClass.OrSelector.newBuilder()
+                                        .addAllSelectors([selectLabel1Selector, selectLabel2Selector]).build()
+                        )
+                        .build()
+        )
+        then:
+        act instanceof Selector.OrMatcher
+        with((Selector.OrMatcher)act) {
+            matchers.size() == 2
+            with((Selector.LabelMatcher)matchers[0]) {
+                name == "foo"
+                values == ["bar"]
+            }
+            with((Selector.LabelMatcher)matchers[1]) {
+                name == "baz"
+                values == ["bar"]
+            }
+        }
+    }
+
+    def "Convert AND match"() {
+        when:
+        def act = Selector.convertToMatcher(
+                BlockchainOuterClass.Selector.newBuilder()
+                        .setAndSelector(
+                                BlockchainOuterClass.AndSelector.newBuilder()
+                                        .addAllSelectors([selectLabel1Selector, selectLabel2Selector]).build()
+                        )
+                        .build()
+        )
+        then:
+        act instanceof Selector.AndMatcher
+        with((Selector.AndMatcher)act) {
+            matchers.size() == 2
+            with((Selector.LabelMatcher)matchers[0]) {
+                name == "foo"
+                values == ["bar"]
+            }
+            with((Selector.LabelMatcher)matchers[1]) {
+                name == "baz"
+                values == ["bar"]
+            }
+        }
+    }
+
+    def "Convert LABEL AND NOT LABEl match"() {
+        setup:
+        def label1 = selectLabel1Selector
+        def label2 = selectLabel2Selector
+        def notLabel2 = BlockchainOuterClass.Selector.newBuilder()
+                .setNotSelector(BlockchainOuterClass.NotSelector.newBuilder().setSelector(label2).build())
+                .build()
+        def and = BlockchainOuterClass.AndSelector.newBuilder()
+                .addAllSelectors([
+                        label1, notLabel2
+                ]).build()
+        when:
+        def act = Selector.convertToMatcher(
+                BlockchainOuterClass.Selector.newBuilder()
+                        .setAndSelector(and)
+                        .build()
+        )
+        then:
+        act instanceof Selector.AndMatcher
+        with((Selector.AndMatcher)act) {
+            matchers.size() == 2
+            with((Selector.LabelMatcher)matchers[0]) {
+                name == "foo"
+                values == ["bar"]
+            }
+            matchers[1] instanceof Selector.NotMatcher
+            with((Selector.NotMatcher)matchers[1]) {
+                matcher instanceof Selector.LabelMatcher
+                with((Selector.LabelMatcher)matcher) {
+                    name == "baz"
+                    values == ["bar"]
+                }
+            }
+        }
+    }
+
+    def "LABEL matches single label"() {
+        setup:
+        def matcher = new Selector.LabelMatcher("test", ["foo"])
+
+        expect:
+        matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
+
+        where:
+        maps << [
+                [test: "foo"],
+                [test: "foo", test2: "bar"],
+                [test2: "foo", test: "foo"],
+        ]
+    }
+
+    def "LABEL matches one label two values"() {
+        setup:
+        def matcher = new Selector.LabelMatcher("test", ["foo", "bar"])
+
+        expect:
+        matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
+
+        where:
+        maps << [
+                [test: "foo"],
+                [test: "foo", test2: "bar"],
+                [test2: "foo", test: "bar"],
+        ]
+    }
+
+    def "AND matches one label"() {
+        setup:
+        def matcher = new Selector.AndMatcher(
+                [
+                new Selector.LabelMatcher("test", ["foo", "bar"])
+                ]
+        )
+
+        expect:
+        matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
+
+        where:
+        maps << [
+                [test: "foo"],
+                [test: "foo", test2: "bar"],
+                [test2: "foo", test: "bar"],
+        ]
+    }
+
+    def "AND matches two labels"() {
+        setup:
+        def matcher = new Selector.AndMatcher(
+                [
+                        new Selector.LabelMatcher("test", ["foo", "bar"]),
+                        new Selector.LabelMatcher("test2", ["baz"])
+                ]
+        )
+
+        expect:
+        matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
+
+        where:
+        maps << [
+                [test: "foo", test2: "baz"],
+                [test: "foo", test3: "bar", test2: "baz"],
+                [test3: "foo", test: "bar", test2: "baz"],
+        ]
+    }
+
+    def "OR matches two labels"() {
+        setup:
+        def matcher = new Selector.OrMatcher(
+                [
+                        new Selector.LabelMatcher("test", ["foo", "bar"]),
+                        new Selector.LabelMatcher("test2", ["baz"])
+                ]
+        )
+
+        expect:
+        matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
+
+        where:
+        maps << [
+                [test: "foo", test2: "baz"],
+                [test: "foo", test3: "bar", test2: "baz"],
+                [test3: "foo", test: "bar", test2: "baz"],
+                [test2: "baz"],
+                [test: "bar"],
+                [test: "foo"],
+        ]
+    }
+
+    def "AND NOT matches two labels"() {
+        setup:
+        def matcher = new Selector.AndMatcher(
+                [
+                        new Selector.LabelMatcher("test", ["foo", "bar"]),
+                        new Selector.NotMatcher(
+                                new Selector.LabelMatcher("test2", ["baz"])
+                        )
+                ]
+        )
+
+        expect:
+        matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
+
+        where:
+        maps << [
+                [test: "foo", test2: "not_baz"],
+                [test: "foo", test3: "bar", test2: "not_baz"],
+                [test: "foo", test3: "bar"],
+                [test3: "foo", test: "bar", test2: "not_baz"],
+                [test3: "foo", test: "bar"],
+                [test: "bar"],
+                [test: "foo"],
+        ]
+    }
+}