diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 08d22006..688ceb55 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -14,6 +14,8 @@ class UpstreamsConfig { @Deprecated("remove it") var quorum: Int = 1 var disableSyncing: Boolean? = null + var disableValidation: Boolean? = null + var minPeers: Int? = 1 set(minPeers) { if (minPeers != null && minPeers < 0) { @@ -29,6 +31,7 @@ class UpstreamsConfig { val copy = Options() copy.disableSyncing = if (this.disableSyncing != null) this.disableSyncing else additional.disableSyncing copy.minPeers = if (this.minPeers != null) this.minPeers else additional.minPeers + copy.disableValidation = if (this.disableValidation != null) this.disableValidation else additional.disableValidation return copy } @@ -37,6 +40,7 @@ class UpstreamsConfig { val options = Options() options.disableSyncing = true options.minPeers = 1 + options.disableValidation = false return options } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 4f158a45..6039219d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -29,15 +29,8 @@ class UpstreamsConfigReader { val defaultOptions = UpstreamsConfig.DefaultOptions() config.defaultOptions.add(defaultOptions) defaultOptions.chains = getListOfString(opts, "chains") - val options = UpstreamsConfig.Options() - defaultOptions.options = options getMapping(opts, "options")?.let { values -> - getValueAsBool(values, "disable-syncing")?.let { - options.disableSyncing = it - } - getValueAsInt(values, "min-peers")?.let { - options.minPeers = it - } + defaultOptions.options = readOptions(values) } } @@ -95,6 +88,7 @@ class UpstreamsConfigReader { internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) { upstream.id = getValueAsString(upNode, "id") + upstream.options = tryReadOptions(upNode) } internal fun readUpstreamGrpc(upNode: MappingNode, upstream: UpstreamsConfig.Upstream) { @@ -122,6 +116,30 @@ class UpstreamsConfigReader { } } + internal fun tryReadOptions(upNode: MappingNode): UpstreamsConfig.Options? { + return if (hasAny(upNode, "options")) { + return getMapping(upNode, "options")?.let { values -> + readOptions(values) + } + } else { + null + } + } + + internal fun readOptions(values: MappingNode): UpstreamsConfig.Options { + val options = UpstreamsConfig.Options() + getValueAsBool(values, "disable-syncing")?.let { + options.disableSyncing = it + } + getValueAsInt(values, "min-peers")?.let { + options.minPeers = it + } + getValueAsBool(values, "disable-validation")?.let { + options.disableValidation = it + } + return options + } + private fun readAuth(authNode: MappingNode?): UpstreamsConfig.Auth? { return getValueAsString(authNode, "type")?.let { return when (it) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index e0e0e114..64bc8796 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -47,13 +47,17 @@ open class ConfiguredUpstreams( val config = readConfig() val defaultOptions = buildDefaultOptions(config) config.upstreams.forEach { up -> - val options = (up.options ?: UpstreamsConfig.Options()) - .merge(UpstreamsConfig.Options.getDefaults()) if (up.connection is UpstreamsConfig.GrpcConnection) { - buildGrpcUpstream(up.connection as UpstreamsConfig.GrpcConnection, options) + buildGrpcUpstream(up.connection as UpstreamsConfig.GrpcConnection) } else { - val chain = chainNames[up.chain] ?: return + val chain = chainNames[up.chain] + if (chain == null) { + log.error("Chain not supported: ${up.chain}") + return@forEach + } + val options = (up.options ?: UpstreamsConfig.Options()) + .merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults()) buildEthereumUpstream(up.connection as UpstreamsConfig.EthereumConnection, chain, options, up.labels) } } @@ -78,19 +82,22 @@ open class ConfiguredUpstreams( private fun buildDefaultOptions(config: UpstreamsConfig): HashMap { val defaultOptions = HashMap() - config.defaultOptions.forEach { df -> - df.chains?.forEach { chainName -> + config.defaultOptions.forEach { defaultsConfig -> + defaultsConfig.chains?.forEach { chainName -> chainNames[chainName]?.let { chain -> - var current = defaultOptions[chain] - if (current == null) { - current = df.options - } else { - current = current.merge(df.options) + defaultsConfig.options?.let { options -> + if (!defaultOptions.containsKey(chain)) { + defaultOptions[chain] = options + } else { + defaultOptions[chain] = defaultOptions[chain]!!.merge(options) + } } - defaultOptions[chain] = current!! } } } + defaultOptions.keys.forEach { chain -> + defaultOptions[chain] = defaultOptions[chain]!!.merge(UpstreamsConfig.Options.getDefaults()) + } return defaultOptions } @@ -129,13 +136,12 @@ open class ConfiguredUpstreams( } } - private fun buildGrpcUpstream(up: UpstreamsConfig.GrpcConnection, options: UpstreamsConfig.Options) { + private fun buildGrpcUpstream(up: UpstreamsConfig.GrpcConnection) { val endpoint = up val ds = GrpcUpstreams( endpoint.host!!, endpoint.port ?: 443, objectMapper, - options, up.auth, this ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 50e94a37..7e61671e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -23,16 +23,20 @@ open class EthereumUpstream( private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) - private val head: EthereumHead = createHead() - - private val validator = UpstreamValidator(this, options) + private val head: EthereumHead = this.createHead() init { log.info("Configured for ${chain.chainName}") api.upstream = this - validator.start() - .subscribe(this::setStatus) + if (options.disableValidation != null && options.disableValidation!!) { + this.setLag(0) + this.setStatus(UpstreamAvailability.OK) + } else { + val validator = UpstreamValidator(this, options) + validator.start() + .subscribe(this::setStatus) + } } open fun createHead(): EthereumHead { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index da3a785c..8093917b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -27,15 +27,13 @@ open class GrpcUpstream( private val chain: Chain, private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, - private val options: UpstreamsConfig.Options, private val targets: CallMethods ): DefaultUpstream() { - 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) + private val options = UpstreamsConfig.Options.getDefaults() private val headBlock = AtomicReference>(null) private val streamBlocks: TopicProcessor> = TopicProcessor.create() private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt index 0de7505b..bdd8488f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt @@ -20,7 +20,6 @@ class GrpcUpstreams( private val host: String, private val port: Int, private val objectMapper: ObjectMapper, - private val options: UpstreamsConfig.Options, private val auth: UpstreamsConfig.TlsAuth? = null, private val upstreams: Upstreams ) { @@ -92,7 +91,7 @@ class GrpcUpstreams( lock.withLock { val current = known[chain] return if (current == null) { - val created = GrpcUpstream(chain, client!!, objectMapper, options, upstreams.targetFor(chain)) + val created = GrpcUpstream(chain, client!!, objectMapper, upstreams.targetFor(chain)) known[chain] = created upstreams.addUpstream(chain, created) created.connect() diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 2623c5bd..11c376aa 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -96,6 +96,22 @@ class UpstreamsConfigReaderSpec extends Specification { } } + def "Parse config with options"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-options.yaml") + when: + def act = reader.read(config) + then: + act != null + act.upstreams.size() == 2 + with(act.upstreams.get(0)) { + options.minPeers == 7 + } + with(act.upstreams.get(1)) { + options.disableValidation == true + } + } + def "Post process for usual strings"() { expect: s == reader.postProcess(s) diff --git a/src/test/resources/upstreams-options.yaml b/src/test/resources/upstreams-options.yaml new file mode 100644 index 00000000..b012e381 --- /dev/null +++ b/src/test/resources/upstreams-options.yaml @@ -0,0 +1,33 @@ +version: v1 + +defaultOptions: + - chains: + - ethereum + options: + disable-syncing: true + min-peers: 3 + +upstreams: + - id: local + chain: ethereum + options: + min-peers: 7 + connection: + ethereum: + rpc: + url: "http://localhost:8545" + ws: + url: "ws://localhost:8546" + origin: "http://localhost" + - id: infura + chain: ethereum + options: + disable-validation: true + connection: + ethereum: + rpc: + url: "https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2" + auth: + type: basic + username: 4fc258fe41a68149c199ad8f281f2015 + password: 1a68f20154fc258fe4149c199ad8f281 \ No newline at end of file