problem: allows multiple different endpoints for same upstream

solution: simplify config, allow to setup only one connection type per upstream

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

@@ -0,0 +1,120 @@
+package io.emeraldpay.dshackle.config
+
+import java.net.URI
+import java.util.*
+import kotlin.collections.ArrayList
+
+class UpstreamsConfig {
+    var version: String? = null
+    var defaultOptions: MutableList<DefaultOptions> = ArrayList<DefaultOptions>()
+    var upstreams: MutableList<Upstream<*>> = ArrayList<Upstream<*>>()
+
+    open class Options {
+        @Deprecated("remove it")
+        var quorum: Int = 1
+        var disableSyncing: Boolean? = null
+        var minPeers: Int? = 1
+            set(minPeers) {
+                if (minPeers != null && minPeers < 0) {
+                    throw IllegalArgumentException("minPeers must be positive number")
+                }
+                field = minPeers
+            }
+
+        fun merge(additional: Options?): Options {
+            if (additional == null) {
+                return this
+            }
+            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
+            return copy
+        }
+
+        companion object {
+            fun getDefaults(): Options {
+                val options = Options()
+                options.disableSyncing = true
+                options.minPeers = 1
+                return options
+            }
+        }
+
+    }
+
+    class DefaultOptions : Options() {
+        var chains: List<String>? = null
+        var options: Options? = null
+    }
+
+    class Upstream<T : UpstreamConnection> {
+        var id: String? = null
+        var chain: String? = null
+        var provider: String? = null
+        var options: Options? = null
+        var isEnabled = true
+        var connection: T? = null
+    }
+
+    open class UpstreamConnection
+
+    class GrpcConnection : UpstreamConnection() {
+        var host: String? = null
+        var port: Int = 0
+        var auth: TlsAuth? = null
+    }
+
+    class EthereumConnection : UpstreamConnection() {
+        var rpc: HttpEndpoint? = null
+        var ws: WsEndpoint? = null
+    }
+
+    class HttpEndpoint(val url: URI) {
+        var auth: Auth? = null
+    }
+
+    class WsEndpoint(val url: URI) {
+        var origin: URI? = null
+    }
+
+    open class Auth {
+        var type: String? = null
+    }
+
+    class BasicAuth : Auth() {
+        var key: String? = null
+    }
+
+    class TlsAuth : Auth() {
+        var ca: String? = null
+        var certificate: String? = null
+        var key: String? = null
+    }
+
+    enum class UpstreamType private constructor(vararg code: String) {
+        ETHEREUM_JSON_RPC("ethereum"),
+        DSHACKLE("dshackle", "grpc"),
+        UNKNOWN("unknown");
+
+        private val code: Array<String>
+
+        init {
+            this.code = code as Array<String>
+            Arrays.sort(this.code)
+        }
+
+        companion object {
+
+            fun byName(code: String): UpstreamType {
+                var code = code
+                code = code.toLowerCase()
+                for (t in UpstreamType.values()) {
+                    if (Arrays.binarySearch(t.code, code) >= 0) {
+                        return t
+                    }
+                }
+                return UNKNOWN
+            }
+        }
+    }
+}

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

@@ -1,17 +1,198 @@
 package io.emeraldpay.dshackle.config
 
+import org.slf4j.LoggerFactory
 import org.yaml.snakeyaml.Yaml
+import org.yaml.snakeyaml.nodes.CollectionNode
+import org.yaml.snakeyaml.nodes.MappingNode
+import org.yaml.snakeyaml.nodes.Node
+import org.yaml.snakeyaml.nodes.ScalarNode
 import java.io.InputStream
+import java.io.InputStreamReader
+import java.lang.IllegalArgumentException
+import java.net.URI
 
 class UpstreamsConfigReader {
 
+    private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
+
     fun read(input: InputStream): UpstreamsConfig {
         val yaml = Yaml()
-        yaml.addTypeDescription(UpstreamsConfig.EndpointTypeYaml())
-        yaml.addTypeDescription(UpstreamsConfig.OptionsYaml())
-        yaml.addTypeDescription(UpstreamsConfig.AuthYaml())
-        return yaml.loadAs(input, UpstreamsConfig::class.java)
+        val configNode = asMappingNode(yaml.compose(InputStreamReader(input)))
+
+        val config = UpstreamsConfig()
+        config.version = getValueAsString(configNode, "version")
+
+        getList<MappingNode>(configNode, "defaultOptions")?.value?.forEach { opts ->
+            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
+                }
+            }
+        }
+
+        config.upstreams = ArrayList<UpstreamsConfig.Upstream<*>>()
+        getList<MappingNode>(configNode, "upstreams")?.value?.forEach { upNode ->
+            val connNode = getMapping(upNode, "connection")
+            if (hasAny(connNode, "ethereum")) {
+                val connConfigNode = getMapping(connNode, "ethereum")!!
+                val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>()
+                upstream.id = getValueAsString(upNode, "id")
+                upstream.provider = getValueAsString(upNode, "provider")
+                upstream.chain = getValueAsString(upNode, "chain")
+                config.upstreams.add(upstream)
+                val connection = UpstreamsConfig.EthereumConnection()
+                upstream.connection = connection
+                getMapping(connConfigNode, "rpc")?.let { node ->
+                    getValueAsString(node, "url")?.let { url ->
+                        val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                        connection.rpc = http
+                        http.auth = readAuth(getMapping(node, "auth"))
+                    }
+                }
+                getMapping(connConfigNode, "ws")?.let { node ->
+                    getValueAsString(node, "url")?.let { url ->
+                        val ws = UpstreamsConfig.WsEndpoint(URI(url))
+                        connection.ws = ws
+                        getValueAsString(node, "origin")?.let { origin ->
+                            ws.origin = URI(origin)
+                        }
+//                    ws.auth = readAuth(getMapping(node, "auth"))
+                    }
+                }
+            } else if (hasAny(connNode, "grpc")) {
+                val connConfigNode = getMapping(connNode, "grpc")!!
+                val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>()
+                upstream.id = getValueAsString(upNode, "id")
+                upstream.provider = getValueAsString(upNode, "provider")
+                config.upstreams.add(upstream)
+                val connection = UpstreamsConfig.GrpcConnection()
+                upstream.connection = connection
+                getValueAsString(connConfigNode, "host")?.let {
+                    connection.host = it
+                }
+                getValueAsInt(connConfigNode, "port")?.let {
+                    connection.port = it
+                }
+                connection.auth = readAuth(getMapping(connConfigNode, "auth")) as UpstreamsConfig.TlsAuth?
+            }
+        }
+
+        return config
     }
 
+    private fun readAuth(authNode: MappingNode?): UpstreamsConfig.Auth? {
+        return getValueAsString(authNode, "type")?.let {
+            return when (it) {
+                "tls" -> {
+                    val auth = UpstreamsConfig.TlsAuth()
+                    auth.ca = getValueAsString(authNode, "ca")
+                    auth.certificate = getValueAsString(authNode, "certificate")
+                    auth.key = getValueAsString(authNode, "key")
+                    auth
+                }
+                "basic" -> {
+                    val auth = UpstreamsConfig.BasicAuth()
+                    auth.key = getValueAsString(authNode, "key")
+                    auth
+                }
+                else -> {
+                    log.warn("Invalid Auth type: $it")
+                    null
+                }
+            }
+        }
+    }
 
+    private fun hasAny(mappingNode: MappingNode?, key: String): Boolean {
+        if (mappingNode == null) {
+            return false
+        }
+        return mappingNode.value
+                .stream()
+                .filter { n -> n.keyNode is ScalarNode }
+                .filter { n ->
+                    val sn = n.keyNode as ScalarNode
+                    key == sn.value
+                }.count() > 0
+    }
+
+    private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? {
+        if (mappingNode == null) {
+            return null
+        }
+        return mappingNode.value
+                .stream()
+                .filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) }
+                .filter { n ->
+                    val sn = n.keyNode as ScalarNode
+                    key == sn.value
+                }
+                .map { n -> n.valueNode as T }
+                .findFirst().let {
+                    if (it.isPresent) {
+                        it.get()
+                    } else {
+                        null
+                    }
+                }
+    }
+
+    private fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? {
+        return getValue(mappingNode, key, MappingNode::class.java)
+    }
+
+    private fun getValue(mappingNode: MappingNode?, key: String): ScalarNode? {
+        return getValue(mappingNode, key, ScalarNode::class.java)
+    }
+
+    private fun <T> getList(mappingNode: MappingNode?, key: String): CollectionNode<T>? {
+        return getValue(mappingNode, key, CollectionNode::class.java) as CollectionNode<T>
+    }
+
+    private fun getListOfString(mappingNode: MappingNode?, key: String): List<String>? {
+        return getList<ScalarNode>(mappingNode, key)?.value
+                ?.map { it.value }
+    }
+
+    private fun getValueAsString(mappingNode: MappingNode?, key: String): String? {
+        return getValue(mappingNode, key)?.let {
+             return@let it.value
+        }
+    }
+
+    private fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? {
+        return getValue(mappingNode, key)?.let {
+            return@let if (it.isPlain) {
+                it.value.toIntOrNull()
+            } else {
+                null
+            }
+        }
+    }
+
+    private fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? {
+        return getValue(mappingNode, key)?.let {
+            return@let if (it.isPlain) {
+                it.value?.toLowerCase() == "true"
+            } else {
+                null
+            }
+        }
+    }
+
+    private fun asMappingNode(node: Node): MappingNode {
+        return if (MappingNode::class.java.isAssignableFrom(node.javaClass)) {
+            node as MappingNode
+        } else {
+            throw IllegalArgumentException("Not a map")
+        }
+    }
 }

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

@@ -41,10 +41,14 @@ 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.provider == "dshackle") {
-                buildGrpcUpstream(up)
+                buildGrpcUpstream(up.connection as UpstreamsConfig.GrpcConnection, options)
             } else {
-                buildEthereumUpstream(up, defaultOptions)
+                val chain = chainNames[up.chain] ?: return
+                buildEthereumUpstream(up.connection as UpstreamsConfig.EthereumConnection, chain, options)
             }
         }
     }
@@ -69,7 +73,7 @@ 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 ->
+            df.chains?.forEach { chainName ->
                 chainNames[chainName]?.let { chain ->
                     var current = defaultOptions[chain]
                     if (current == null) {
@@ -77,56 +81,45 @@ open class ConfiguredUpstreams(
                     } else {
                         current = current.merge(df.options)
                     }
-                    defaultOptions[chain] = current
+                    defaultOptions[chain] = current!!
                 }
             }
         }
         return defaultOptions
     }
 
-    private fun buildEthereumUpstream(up: UpstreamsConfig.Upstream,
-                                      defaultOptions: HashMap<Chain, UpstreamsConfig.Options>) {
-        val chain = chainNames[up.chain] ?: return
+    private fun buildEthereumUpstream(up: UpstreamsConfig.EthereumConnection,
+                                      chain: Chain,
+                                      options: UpstreamsConfig.Options) {
         var rpcApi: EthereumApi? = null
         var wsApi: EthereumWs? = null
         val urls = ArrayList<URI>()
-        up.endpoints.forEach { endpoint ->
-            if (endpoint.type == UpstreamsConfig.EndpointType.JSON_RPC) {
-                rpcApi = EthereumApi(
-                        DefaultRpcClient(DefaultRpcTransport(endpoint.url)),
-                        objectMapper,
-                        chain
-                )
-            }
-            if (endpoint.type == UpstreamsConfig.EndpointType.WEBSOCKET) {
-                wsApi = EthereumWs(
-                        endpoint.url,
-                        endpoint.origin ?: URI("http://localhost")
-                )
-                wsApi!!.connect()
-            }
+        up.rpc?.let { endpoint ->
+            rpcApi = EthereumApi(
+                    DefaultRpcClient(DefaultRpcTransport(endpoint.url)),
+                    objectMapper,
+                    chain
+            )
+            urls.add(endpoint.url)
+        }
+        up.ws?.let { endpoint ->
+            wsApi = EthereumWs(
+                    endpoint.url,
+                    endpoint.origin ?: URI("http://localhost")
+            )
+            wsApi!!.connect()
             urls.add(endpoint.url)
         }
-        val options = (up.options ?: UpstreamsConfig.Options())
-                .merge(defaultOptions[chain])
-                .merge(UpstreamsConfig.Options.getDefaults())
         if (rpcApi != null) {
             log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
             getOrCreateUpstream(chain).addUpstream(EthereumUpstream(chain, rpcApi!!, wsApi, options))
         }
     }
 
-    private fun buildGrpcUpstream(up: UpstreamsConfig.Upstream) {
-        if (up.endpoints.size == 0) {
-            return
-        }
-        val options = (up.options ?: UpstreamsConfig.Options())
-                .merge(UpstreamsConfig.Options.getDefaults())
-
-        val endpoint = up.endpoints.first()
-        if (endpoint.type == UpstreamsConfig.EndpointType.DSHACKLE) {
+    private fun buildGrpcUpstream(up: UpstreamsConfig.GrpcConnection, options: UpstreamsConfig.Options) {
+        val endpoint = up
             val ds = GrpcUpstreams(
-                    endpoint.host,
+                    endpoint.host!!,
                     endpoint.port ?: 443,
                     objectMapper,
                     options
@@ -140,7 +133,6 @@ open class ConfiguredUpstreams(
                         log.info("Subscribed to $it through gRPC at ${endpoint.host}:${endpoint.port}")
                         getOrCreateUpstream(it).addUpstream(ds.getOrCreate(it))
                     }
-        }
     }
 
     override fun getUpstream(chain: Chain): AggregatedUpstreams? {

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

@@ -21,7 +21,7 @@ class UpstreamValidator(
             if (syncing.get().isSyncing) {
                 return UpstreamAvailability.SYNCING
             }
-            if (peerCount.get() < options.minPeers) {
+            if (options.minPeers != null && peerCount.get() < options.minPeers!!) {
                 return UpstreamAvailability.IMMATURE
             }
             return UpstreamAvailability.OK