diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index bf261a9c..0fcc2eaa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/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() { + + companion object { + @JvmStatic fun fromMap(map: Map): Labels { + val labels = Labels() + map.entries.forEach() { kv -> + labels.put(kv.key, kv.value) + } + return labels + } + } } enum class UpstreamType private constructor(vararg code: String) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 35cfa4d2..2b7f89d3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/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 upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable") + val matcher = Selector.convertToMatcher(request.selector) + 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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index 327e76a2..0ca30803 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/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 { 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 - } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index f1ad51c9..920bd37b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/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 { 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 { 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) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt index f6018cec..9a85f21b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt @@ -11,7 +11,7 @@ abstract class AggregatedUpstreams: Upstream { abstract fun getAll(): List abstract fun addUpstream(upstream: Upstream) - abstract fun getApis(quorum: Int): Iterator + abstract fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator override fun observeStatus(): Flux { 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 { - return getAll().any { it.isAvailable() } + override fun isAvailable(matcher: Selector.Matcher): Boolean { + 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() - options.quorum = getAll().filter { - it.getStatus() == UpstreamAvailability.OK - }.sumBy { - it.getOptions().quorum - } - return options - } - - class SingleApi( - private val quorumApi: QuorumApi - ): Iterator { - - 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, - private val quorum: Int, - private var pos: Int - ): Iterator { - - 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") - } + return UpstreamsConfig.Options() } class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index cdf3d019..de4ab552 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/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 { + override fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator { 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 { - return getApis(1).next() + override fun getApi(matcher: Selector.Matcher): EthereumApi { + return getApis(1, matcher).next() } override fun getHead(): EthereumHead { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index d3468629..e40fc3fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt index 89844ce9..7c53467e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt +++ b/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>): CompletableFuture { 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)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 32be2967..73a69792 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/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 { - return status.get() == UpstreamAvailability.OK + override fun isAvailable(matcher: Selector.Matcher): Boolean { + 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 { 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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt new file mode 100644 index 00000000..ac3c9ff0 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt @@ -0,0 +1,27 @@ +package io.emeraldpay.dshackle.upstream + +class FilteringApiIterator( + private val apis: List, + private val quorum: Int, + private var pos: Int, + private val matcher: Selector.Matcher +): Iterator { + + 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") + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index 22fde950..cb5c8b36 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -41,14 +41,13 @@ open class GrpcUpstream( private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE) private val nodes = AtomicReference(NodeDetailsList()) private val head = Head(this) - private val api: EthereumApi private val statusStream: TopicProcessor = TopicProcessor.create() private val supportedMethods = HashSet() + private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper) - init { - val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper) - val rpcClient = DefaultRpcClient(grpcTransport) - api = EthereumApi(rpcClient, objectMapper, chain) + open fun createApi(matcher: Selector.Matcher): EthereumApi { + val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher)) + return 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 { - return headBlock.get() != null + override fun isAvailable(matcher: Selector.Matcher): Boolean { + 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 { - return api + override fun getApi(matcher: Selector.Matcher): EthereumApi { + return createApi(matcher) } override fun getOptions(): UpstreamsConfig.Options { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt new file mode 100644 index 00000000..d78614ec --- /dev/null +++ b/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): 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 { + 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 { + 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() + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 1525dd4e..3c9e289e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/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 fun getHead(): EthereumHead - fun getApi(): EthereumApi + fun getApi(matcher: Selector.Matcher): EthereumApi fun getOptions(): UpstreamsConfig.Options fun getSupportedTargets(): Set } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt index 11fe4ffc..fc67cd63 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt +++ b/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 } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy index 4842c540..cad8e867 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy @@ -23,7 +23,7 @@ import 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)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy index 03b62eda..cc38d7f0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy +++ b/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) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index a42f76d5..145fc802 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/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 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy new file mode 100644 index 00000000..74cc9767 --- /dev/null +++ b/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() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy new file mode 100644 index 00000000..b961dd59 --- /dev/null +++ b/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"], + ] + } +}