StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: allows to disable upstream validation

Browse Source
This commit is contained in:
Igor Artamonov
2019-08-13 16:33:44 -04:00
parent 9f211232fd
commit 3957b8cf42
8 changed files with 110 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -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
}
}

34
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -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<UpstreamsConfig.GrpcConnection>) {
@@ -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) {

34
src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt
View File

@@ -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<Chain, UpstreamsConfig.Options> {
val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
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
)

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt
View File

@@ -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 {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt
View File

@@ -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<BlockJson<TransactionId>>(null)
private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
private val status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt
View File

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

16
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

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

33
src/test/resources/upstreams-options.yaml Normal file
View File

@@ -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