diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt new file mode 100644 index 00000000..ce51f3f9 --- /dev/null +++ b/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 + ) { + + 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 { + val i = seq++ + if (seq >= Int.MAX_VALUE / 2) { + seq = 0 + } + return QuorumApi(upstreams, 1, seq) + } + + class SingleApi( + val quorumApi: QuorumApi + ): Iterator { + + 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, + val quorum: Int, + var pos: Int + ): Iterator { + + 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") + } + + } + + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt new file mode 100644 index 00000000..d65aec99 --- /dev/null +++ b/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 { + + private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java) + private val flux: Flux> + private val head = AtomicReference>(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> { + val curr = head.get() + if (curr != null) { + return Mono.just(curr) + } + return Mono.from(getFlux()) + } + + override fun getFlux(): Flux> { + return Flux.from(this.flux) + .onBackpressureLatest() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt index 0cb71a68..522b5523 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt @@ -22,7 +22,7 @@ class EthereumWsHead( } override fun getFlux(): Flux> { - return Flux.from(stream) + return ws.getFlux() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 8835cbab..329ea5c1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/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 + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index 1a730d0d..1d3e4b05 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/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>() + private val chainMapping = HashMap() 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>() + 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? { - val list = 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) + fun ethereumUpstream(chain: Chain): ChainConnect? { + return chainMapping[chain] } } \ No newline at end of file