solution: group all upstreams per chain

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

@@ -0,0 +1,75 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.grpc.Chain
+import java.lang.IllegalStateException
+
+class ChainConnect(
+        val chain: Chain,
+        val upstreams: List<Upstream>
+        ) {
+
+    private var seq = 0
+
+    val head: EthereumHead = if (upstreams.size == 1) {
+        upstreams.first().head
+    } else {
+        EthereumHeadMerge(upstreams.map { it.head })
+    }
+
+    val api: EthereumApi
+        get() {
+            return getApis(1).next()
+        }
+
+    fun getApis(quorum: Int): Iterator<EthereumApi> {
+        val i = seq++
+        if (seq >= Int.MAX_VALUE / 2) {
+            seq = 0
+        }
+        return QuorumApi(upstreams, 1, seq)
+    }
+
+    class SingleApi(
+            val quorumApi: QuorumApi
+    ): Iterator<EthereumApi> {
+
+        var consumed = false
+
+        override fun hasNext(): Boolean {
+            return !consumed && quorumApi.hasNext()
+        }
+
+        override fun next(): EthereumApi {
+            consumed = true
+            return quorumApi.next()
+        }
+    }
+
+    class QuorumApi(
+            val apis: List<Upstream>,
+            val quorum: Int,
+            var pos: Int
+    ): Iterator<EthereumApi> {
+
+        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++]
+                if (api.isAvailable()) {
+                    consumed++
+                    return api.api
+                }
+            }
+            throw IllegalStateException("No upstream API available")
+        }
+
+    }
+
+
+}

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

@@ -0,0 +1,46 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+import java.util.concurrent.atomic.AtomicReference
+
+class EthereumHeadMerge(
+        private val upstreams: List<EthereumHead>
+): EthereumHead {
+
+    private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java)
+    private val flux: Flux<BlockJson<TransactionId>>
+    private val head = AtomicReference<BlockJson<TransactionId>>(null)
+
+    init {
+        val fluxes = upstreams.map { it.getFlux() }
+        flux = Flux.merge(fluxes)
+                .filter {
+                    val curr = head.get()
+                    curr == null || curr.totalDifficulty < it.totalDifficulty
+                }
+                .publish()
+                .autoConnect()
+
+        Flux.from(flux).subscribe {
+            head.set(it)
+        }
+    }
+
+    override fun getHead(): Mono<BlockJson<TransactionId>> {
+        val curr = head.get()
+        if (curr != null) {
+            return Mono.just(curr)
+        }
+        return Mono.from(getFlux())
+    }
+
+    override fun getFlux(): Flux<BlockJson<TransactionId>> {
+        return Flux.from(this.flux)
+                .onBackpressureLatest()
+    }
+}

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

@@ -22,7 +22,7 @@ class EthereumWsHead(
     }
 
     override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return Flux.from(stream)
+        return ws.getFlux()
     }
 
 }

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

@@ -24,4 +24,8 @@ class Upstream(
     init {
         log.info("Configured for ${chain.chainName}")
     }
+
+    fun isAvailable(): Boolean {
+        return true
+    }
 }

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

@@ -21,12 +21,13 @@ class Upstreams(
 ) {
 
     private val log = LoggerFactory.getLogger(Upstreams::class.java)
-    private var seq = 0
+    private val chainMapping = HashMap<Chain, ChainConnect>()
-    private val chainMapping = HashMap<Chain, ArrayList<Upstream>>()
 
     private val chainNames = mapOf(
             "ethereum" to Chain.ETHEREUM,
             "ethereum-classic" to Chain.ETHEREUM_CLASSIC,
+            "eth" to Chain.ETHEREUM,
+            "etc" to Chain.ETHEREUM_CLASSIC,
             "morden" to Chain.MORDEN
     )
 
@@ -47,6 +48,8 @@ class Upstreams(
         val reader = UpstreamsReader()
         val config = reader.read(upstreamConfig.inputStream())
 
+        val groups = HashMap<Chain, ArrayList<Upstream>>()
+
         config.upstreams.forEach { up ->
             val chain = chainNames[up.chain] ?: return@forEach
             var rpcApi: EthereumApi? = null
@@ -65,27 +68,23 @@ class Upstreams(
                             endpoint.url,
                             endpoint.origin ?: URI("http://localhost")
                     )
+                    wsApi!!.connect()
                 }
                 urls.add(endpoint.url)
             }
             if (rpcApi != null) {
                 log.info("Info using ${chain.chainName} upstream, at ${urls.joinToString()}")
-                val current = chainMapping[chain] ?: ArrayList()
+                val current = groups[chain] ?: ArrayList()
                 current.add(Upstream(chain, rpcApi!!, wsApi))
-                chainMapping[chain] = current
+                groups[chain] = current
             }
         }
+        groups.forEach { chain, group ->
+            chainMapping[chain] = ChainConnect(chain, group)
+        }
     }
 
-    fun ethereumUpstream(chain: Chain): Upstream? {
+    fun ethereumUpstream(chain: Chain): ChainConnect? {
-        val list = chainMapping[chain]
+        return chainMapping[chain]
-        if (list == null || list.isEmpty()) {
-            return null
-        }
-        val i = seq++
-        if (seq >= Int.MAX_VALUE / 2) {
-            seq = 0
-        }
-        return list.get(i % list.size)
     }
 }