better solution

This commit is contained in:
Maxksim Fomenkov
2022-09-29 02:20:26 +03:00
parent f4ff5ab046
commit 69f1f18a5a
5 changed files with 22 additions and 6 deletions

View File

@@ -15,6 +15,7 @@
*/ */
package io.emeraldpay.dshackle package io.emeraldpay.dshackle
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
/** /**
@@ -30,4 +31,6 @@ open class SilentException(message: String) : Exception(message) {
} }
class DataUnavailable(val code: String) : SilentException("Data is unavailable: $code") class DataUnavailable(val code: String) : SilentException("Data is unavailable: $code")
class NoMatchingUpstream(matcher: Selector.LabelSelectorMatcher) : SilentException("No configured upstream matching selector [${matcher.describeInternal()}]")
} }

View File

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.quorum.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.dshackle.upstream.MultistreamHolder
@@ -51,6 +52,7 @@ import java.util.EnumMap
@Service @Service
open class NativeCall( open class NativeCall(
@Autowired private val multistreamHolder: MultistreamHolder, @Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val configuredUpstreams: ConfiguredUpstreams,
@Autowired private val signer: ResponseSigner @Autowired private val signer: ResponseSigner
) { ) {
@@ -150,6 +152,11 @@ open class NativeCall(
return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number)))
} }
val matcher = Selector.convertToMatcher(request.selector)
if (!configuredUpstreams.hasMatchingUpstream(chain, matcher)) {
return Flux.error(CallFailure(0, SilentException.NoMatchingUpstream(matcher)))
}
val upstream = multistreamHolder.getUpstream(chain) val upstream = multistreamHolder.getUpstream(chain)
?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain))) ?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain)))

View File

@@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HttpRpcFactory import io.emeraldpay.dshackle.upstream.HttpRpcFactory
import io.emeraldpay.dshackle.upstream.MergedHead import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
@@ -107,6 +108,11 @@ open class ConfiguredUpstreams(
} }
} }
fun hasMatchingUpstream(chain: Chain, matcher: Selector.LabelSelectorMatcher): Boolean =
config.upstreams.any { up ->
up.chain?.equals(chain.chainName, ignoreCase = true) ?: true && matcher.matches(up.labels)
}
private fun buildDefaultOptions(config: UpstreamsConfig): HashMap<Chain, UpstreamsConfig.Options> { private fun buildDefaultOptions(config: UpstreamsConfig): HashMap<Chain, UpstreamsConfig.Options> {
val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>() val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
config.defaultOptions.forEach { defaultsConfig -> config.defaultOptions.forEach { defaultsConfig ->

View File

@@ -156,10 +156,6 @@ class FilteredApis(
} }
override fun subscribe(subscriber: Subscriber<in Upstream>) { override fun subscribe(subscriber: Subscriber<in Upstream>) {
if (allUpstreams.none { matcher.matches(it) }) {
Flux.empty<Upstream>().subscribe(subscriber)
return
}
// initially try only standard upstreams // initially try only standard upstreams
val first = Flux.fromIterable(primaryUpstreams) val first = Flux.fromIterable(primaryUpstreams)
val second = Flux.fromIterable(secondaryUpstreams) val second = Flux.fromIterable(secondaryUpstreams)

View File

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.quorum.QuorumRpcReader
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.test.MultistreamHolderMock
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.AlwaysQuorum
@@ -51,14 +52,17 @@ class NativeCallSpec extends Specification {
ObjectMapper objectMapper = Global.objectMapper ObjectMapper objectMapper = Global.objectMapper
def nativeCall(MultistreamHolder upstreams = null, ResponseSigner signer = null) { def nativeCall(MultistreamHolder upstreams = null, ResponseSigner signer = null, ConfiguredUpstreams configuredUpstreams = null) {
if (upstreams == null) { if (upstreams == null) {
upstreams = Stub(MultistreamHolder) upstreams = Stub(MultistreamHolder)
} }
if (signer == null) { if (signer == null) {
signer = Stub(ResponseSigner) signer = Stub(ResponseSigner)
} }
new NativeCall(upstreams, signer) if (configuredUpstreams == null) {
configuredUpstreams = Stub(ConfiguredUpstreams)
}
new NativeCall(upstreams, configuredUpstreams, signer)
} }
def "Tries router first"() { def "Tries router first"() {