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

problem: prints status of chains that are not configured

Browse Source
This commit is contained in:
Igor Artamonov
2019-07-25 23:58:48 -04:00
parent 144008e5aa
commit 7467e19ea0
7 changed files with 21 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
View File

@@ -17,7 +17,7 @@ class Describe(
fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver<BlockchainOuterClass.DescribeResponse>) { fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver<BlockchainOuterClass.DescribeResponse>) {
val resp = BlockchainOuterClass.DescribeResponse.newBuilder() val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
upstreams.getAvailable().forEach { chain -> upstreams.getAvailable().forEach { chain ->
upstreams.ethereumUpstream(chain).let { chainUpstreams -> upstreams.getUpstream(chain)?.let { chainUpstreams ->
val quorum = chainUpstreams.getAll().map { u -> val quorum = chainUpstreams.getAll().map { u ->
if (u.getStatus() == UpstreamAvailability.OK) { if (u.getStatus() == UpstreamAvailability.OK) {
u.getOptions().quorum u.getOptions().quorum

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -28,7 +28,7 @@ class NativeCall(
if (chain == Chain.UNSPECIFIED) { if (chain == Chain.UNSPECIFIED) {
throw Exception("Invalid chain id: ${request.chain.number}") 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() request.itemsList.toFlux()
.map { .map {
val method = it.target val method = it.target

6
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -28,7 +28,7 @@ class StreamHead(
@PostConstruct @PostConstruct
fun init() { fun init() {
listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_MORDEN, Chain.TESTNET_KOVAN).forEach { chain -> 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() clients[chain] = ConcurrentLinkedQueue()
subscribe(chain) subscribe(chain)
} }
@@ -36,7 +36,7 @@ class StreamHead(
} }
private fun subscribe(chain: Chain) { private fun subscribe(chain: Chain) {
upstreams.ethereumUpstream(chain)!!.getHead().getFlux() upstreams.getUpstream(chain)!!.getHead().getFlux()
.doOnComplete { .doOnComplete {
log.info("Closing streams for ${chain.chainCode}") log.info("Closing streams for ${chain.chainCode}")
clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client -> clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client ->
@@ -67,7 +67,7 @@ class StreamHead(
} }
fun process(chain: Chain, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean { fun process(chain: Chain, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
val upstream = upstreams.ethereumUpstream(chain) ?: return false val upstream = upstreams.getUpstream(chain) ?: return false
val head = upstream.getHead().getHead() val head = upstream.getHead().getHead()
return head.map { return head.map {
notify(chain, it, client) notify(chain, it, client)

4
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
View File

@@ -37,7 +37,7 @@ class TrackAddress(
fun init() { fun init() {
allChains.forEach { chain -> allChains.forEach { chain ->
clients[chain] = ConcurrentLinkedQueue() clients[chain] = ConcurrentLinkedQueue()
upstreams.ethereumUpstream(chain)?.getHead()?.let { head -> upstreams.getUpstream(chain)?.getHead()?.let { head ->
head.getFlux().subscribe { verifyAll(chain) } head.getFlux().subscribe { verifyAll(chain) }
} }
} }
@@ -117,7 +117,7 @@ class TrackAddress(
} }
private fun verify(chain: Chain, group: List<TrackedAddress>): Flux<TrackedAddress> { private fun verify(chain: Chain, group: List<TrackedAddress>): Flux<TrackedAddress> {
val up = upstreams.ethereumUpstream(chain) val up = upstreams.getUpstream(chain) ?: return Flux.empty<TrackedAddress>()
return group.toFlux() return group.toFlux()
.flatMap { a -> .flatMap { a ->
up.getApi() up.getApi()

6
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
View File

@@ -35,7 +35,7 @@ class TrackTx(
fun init() { fun init() {
listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain -> listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain ->
clients[chain] = ConcurrentLinkedQueue() clients[chain] = ConcurrentLinkedQueue()
upstreams.ethereumUpstream(chain)?.getHead()?.let { head -> upstreams.getUpstream(chain)?.getHead()?.let { head ->
head.getFlux().subscribe { verifyAll(chain) } head.getFlux().subscribe { verifyAll(chain) }
} }
} }
@@ -61,7 +61,7 @@ class TrackTx(
} }
private fun loadWeight(tx: TrackedTx): Mono<TrackedTx> { private fun loadWeight(tx: TrackedTx): Mono<TrackedTx> {
val upstream = upstreams.ethereumUpstream(tx.chain) val upstream = upstreams.getUpstream(tx.chain)!!
return upstream.getApi() return upstream.getApi()
.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash))
.map { block -> .map { block ->
@@ -81,7 +81,7 @@ class TrackTx(
private fun verify(tx: TrackedTx): Boolean { private fun verify(tx: TrackedTx): Boolean {
val found = tx.status.found val found = tx.status.found
val mined = tx.status.mined val mined = tx.status.mined
val upstream = upstreams.ethereumUpstream(tx.chain) val upstream = upstreams.getUpstream(tx.chain)!!
val execution = upstream.getApi() val execution = upstream.getApi()
.executeAndConvert(Commands.eth().getTransaction(tx.txid)) .executeAndConvert(Commands.eth().getTransaction(tx.txid))
val update = execution.flatMap { val update = execution.flatMap {

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

@@ -40,17 +40,13 @@ open class ConfiguredUpstreams(
fun start() { fun start() {
val config = readConfig() val config = readConfig()
val defaultOptions = buildDefaultOptions(config) val defaultOptions = buildDefaultOptions(config)
val groups = HashMap<Chain, ArrayList<Upstream>>()
config.upstreams.forEach { up -> config.upstreams.forEach { up ->
if (up.provider == "dshackle") { if (up.provider == "dshackle") {
buildGrpcUpstream(up) buildGrpcUpstream(up)
} else { } else {
buildEthereumUpstream(up, defaultOptions, groups) buildEthereumUpstream(up, defaultOptions)
} }
} }
groups.forEach { chain, group ->
chainMapping[chain] = ChainUpstreams(chain, group)
}
} }
private fun readConfig(): UpstreamsConfig { private fun readConfig(): UpstreamsConfig {
@@ -89,8 +85,7 @@ open class ConfiguredUpstreams(
} }
private fun buildEthereumUpstream(up: UpstreamsConfig.Upstream, private fun buildEthereumUpstream(up: UpstreamsConfig.Upstream,
defaultOptions: HashMap<Chain, UpstreamsConfig.Options>, defaultOptions: HashMap<Chain, UpstreamsConfig.Options>) {
groups: HashMap<Chain, ArrayList<Upstream>>) {
val chain = chainNames[up.chain] ?: return val chain = chainNames[up.chain] ?: return
var rpcApi: EthereumApi? = null var rpcApi: EthereumApi? = null
var wsApi: EthereumWs? = null var wsApi: EthereumWs? = null
@@ -117,9 +112,7 @@ open class ConfiguredUpstreams(
.merge(UpstreamsConfig.Options.getDefaults()) .merge(UpstreamsConfig.Options.getDefaults())
if (rpcApi != null) { if (rpcApi != null) {
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val current = groups[chain] ?: ArrayList() getOrCreateUpstream(chain).addUpstream(EthereumUpstream(chain, rpcApi!!, wsApi, options))
current.add(EthereumUpstream(chain, rpcApi!!, wsApi, options))
groups[chain] = current
} }
} }
@@ -145,12 +138,16 @@ open class ConfiguredUpstreams(
} }
.subscribe { .subscribe {
log.info("Subscribed to $it through gRPC at ${endpoint.host}:${endpoint.port}") 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] val current = chainMapping[chain]
if (current == null) { if (current == null) {
val created = ChainUpstreams(chain, ArrayList<Upstream>()) val created = ChainUpstreams(chain, ArrayList<Upstream>())

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

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
interface Upstreams { interface Upstreams {
fun ethereumUpstream(chain: Chain): AggregatedUpstreams fun getOrCreateUpstream(chain: Chain): AggregatedUpstreams
fun getUpstream(chain: Chain): AggregatedUpstreams?
fun getAvailable(): List<Chain> fun getAvailable(): List<Chain>
} }