From 7467e19ea0032aefab2d1caed6994e6c041364c0 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 25 Jul 2019 23:58:48 -0400 Subject: [PATCH] problem: prints status of chains that are not configured --- .../io/emeraldpay/dshackle/rpc/Describe.kt | 2 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 2 +- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 6 +++--- .../emeraldpay/dshackle/rpc/TrackAddress.kt | 4 ++-- .../io/emeraldpay/dshackle/rpc/TrackTx.kt | 6 +++--- .../dshackle/upstream/ConfiguredUpstreams.kt | 21 ++++++++----------- .../emeraldpay/dshackle/upstream/Upstreams.kt | 3 ++- 7 files changed, 21 insertions(+), 23 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index cb2bed9b..a1b93430 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -17,7 +17,7 @@ class Describe( fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver) { val resp = BlockchainOuterClass.DescribeResponse.newBuilder() upstreams.getAvailable().forEach { chain -> - upstreams.ethereumUpstream(chain).let { chainUpstreams -> + upstreams.getUpstream(chain)?.let { chainUpstreams -> val quorum = chainUpstreams.getAll().map { u -> if (u.getStatus() == UpstreamAvailability.OK) { u.getOptions().quorum diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index bbd09ac2..81699491 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -28,7 +28,7 @@ class NativeCall( if (chain == Chain.UNSPECIFIED) { throw Exception("Invalid chain id: ${request.chain.number}") } - val upstream = upstreams.ethereumUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable") + val upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable") request.itemsList.toFlux() .map { val method = it.target diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 559841e9..3f7769e8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -28,7 +28,7 @@ class StreamHead( @PostConstruct fun init() { listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_MORDEN, Chain.TESTNET_KOVAN).forEach { chain -> - if (upstreams.ethereumUpstream(chain)?.getHead() != null) { + if (upstreams.getUpstream(chain)?.getHead() != null) { clients[chain] = ConcurrentLinkedQueue() subscribe(chain) } @@ -36,7 +36,7 @@ class StreamHead( } private fun subscribe(chain: Chain) { - upstreams.ethereumUpstream(chain)!!.getHead().getFlux() + upstreams.getUpstream(chain)!!.getHead().getFlux() .doOnComplete { log.info("Closing streams for ${chain.chainCode}") clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client -> @@ -67,7 +67,7 @@ class StreamHead( } fun process(chain: Chain, client: StreamSender): Boolean { - val upstream = upstreams.ethereumUpstream(chain) ?: return false + val upstream = upstreams.getUpstream(chain) ?: return false val head = upstream.getHead().getHead() return head.map { notify(chain, it, client) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index be5ab808..8d237fc2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -37,7 +37,7 @@ class TrackAddress( fun init() { allChains.forEach { chain -> clients[chain] = ConcurrentLinkedQueue() - upstreams.ethereumUpstream(chain)?.getHead()?.let { head -> + upstreams.getUpstream(chain)?.getHead()?.let { head -> head.getFlux().subscribe { verifyAll(chain) } } } @@ -117,7 +117,7 @@ class TrackAddress( } private fun verify(chain: Chain, group: List): Flux { - val up = upstreams.ethereumUpstream(chain) + val up = upstreams.getUpstream(chain) ?: return Flux.empty() return group.toFlux() .flatMap { a -> up.getApi() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index 110a3263..3a370efa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -35,7 +35,7 @@ class TrackTx( fun init() { listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain -> clients[chain] = ConcurrentLinkedQueue() - upstreams.ethereumUpstream(chain)?.getHead()?.let { head -> + upstreams.getUpstream(chain)?.getHead()?.let { head -> head.getFlux().subscribe { verifyAll(chain) } } } @@ -61,7 +61,7 @@ class TrackTx( } private fun loadWeight(tx: TrackedTx): Mono { - val upstream = upstreams.ethereumUpstream(tx.chain) + val upstream = upstreams.getUpstream(tx.chain)!! return upstream.getApi() .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) .map { block -> @@ -81,7 +81,7 @@ class TrackTx( private fun verify(tx: TrackedTx): Boolean { val found = tx.status.found val mined = tx.status.mined - val upstream = upstreams.ethereumUpstream(tx.chain) + val upstream = upstreams.getUpstream(tx.chain)!! val execution = upstream.getApi() .executeAndConvert(Commands.eth().getTransaction(tx.txid)) val update = execution.flatMap { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index a9776d9a..a28ed540 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -40,17 +40,13 @@ open class ConfiguredUpstreams( fun start() { val config = readConfig() val defaultOptions = buildDefaultOptions(config) - val groups = HashMap>() config.upstreams.forEach { up -> if (up.provider == "dshackle") { buildGrpcUpstream(up) } else { - buildEthereumUpstream(up, defaultOptions, groups) + buildEthereumUpstream(up, defaultOptions) } } - groups.forEach { chain, group -> - chainMapping[chain] = ChainUpstreams(chain, group) - } } private fun readConfig(): UpstreamsConfig { @@ -89,8 +85,7 @@ open class ConfiguredUpstreams( } private fun buildEthereumUpstream(up: UpstreamsConfig.Upstream, - defaultOptions: HashMap, - groups: HashMap>) { + defaultOptions: HashMap) { val chain = chainNames[up.chain] ?: return var rpcApi: EthereumApi? = null var wsApi: EthereumWs? = null @@ -117,9 +112,7 @@ open class ConfiguredUpstreams( .merge(UpstreamsConfig.Options.getDefaults()) if (rpcApi != null) { log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val current = groups[chain] ?: ArrayList() - current.add(EthereumUpstream(chain, rpcApi!!, wsApi, options)) - groups[chain] = current + getOrCreateUpstream(chain).addUpstream(EthereumUpstream(chain, rpcApi!!, wsApi, options)) } } @@ -145,12 +138,16 @@ open class ConfiguredUpstreams( } .subscribe { log.info("Subscribed to $it through gRPC at ${endpoint.host}:${endpoint.port}") - ethereumUpstream(it).addUpstream(ds.getOrCreate(it)) + getOrCreateUpstream(it).addUpstream(ds.getOrCreate(it)) } } } - override fun ethereumUpstream(chain: Chain): ChainUpstreams { + override fun getUpstream(chain: Chain): AggregatedUpstreams? { + return chainMapping[chain] + } + + override fun getOrCreateUpstream(chain: Chain): ChainUpstreams { val current = chainMapping[chain] if (current == null) { val created = ChainUpstreams(chain, ArrayList()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index 5bec84d0..af44bcc3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.grpc.Chain interface Upstreams { - fun ethereumUpstream(chain: Chain): AggregatedUpstreams + fun getOrCreateUpstream(chain: Chain): AggregatedUpstreams + fun getUpstream(chain: Chain): AggregatedUpstreams? fun getAvailable(): List } \ No newline at end of file