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

problem: unhandled error for second call if first had failed

Browse Source
This commit is contained in:
Igor Artamonov
2019-11-08 18:54:52 -05:00
parent 559a464252
commit 9ca9f4e8ae
2 changed files with 26 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,6 +20,7 @@ import java.time.Duration
class Defaults { class Defaults {
companion object { companion object {
val timeout = Duration.ofSeconds(60) val timeout: Duration = Duration.ofSeconds(60)
val timeoutInternal: Duration = timeout.dividedBy(4)
} }
} }

57
src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt
View File

@@ -18,19 +18,14 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.ReactorBatch
import org.apache.commons.lang3.exception.ExceptionUtils
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers import reactor.core.scheduler.Schedulers
import java.net.ConnectException
import java.time.Duration import java.time.Duration
import java.util.concurrent.Executors import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
class UpstreamValidator( class UpstreamValidator(
private val ethereumUpstream: EthereumUpstream, private val ethereumUpstream: EthereumUpstream,
@@ -42,40 +37,36 @@ class UpstreamValidator(
} }
fun validate(): Mono<UpstreamAvailability> { fun validate(): Mono<UpstreamAvailability> {
val batch = ReactorBatch() return ethereumUpstream
val peerCount = batch.add(Commands.net().peerCount()).result .getApi(Selector.empty)
val syncing = batch.add(Commands.eth().syncing()).result .flatMapMany { api ->
return ethereumUpstream.getApi(Selector.empty) api.rpcClient
.subscribeOn(scheduler) .execute(Commands.eth().syncing())
.flatMapMany { api -> api.rpcClient.execute(batch) } .timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Syncing")))
.timeout(Defaults.timeout, Mono.error(Exception("Validation timeout"))) .flatMap { value ->
.then(syncing) if (value.isSyncing) {
.flatMap { value -> Mono.just(UpstreamAvailability.SYNCING)
if (value.isSyncing) { } else {
Mono.just(UpstreamAvailability.SYNCING) api.rpcClient.execute(Commands.net().peerCount())
} else { .timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Peers")))
peerCount.map { count -> .map { count ->
val minPeers = options.minPeers ?: 1 val minPeers = options.minPeers ?: 1
if (count < minPeers) { if (count < minPeers) {
UpstreamAvailability.IMMATURE UpstreamAvailability.IMMATURE
} else { } else {
UpstreamAvailability.OK UpstreamAvailability.OK
}
}
}
} }
}
}
}
.doOnError { err ->
if (ExceptionUtils.hasCause(err, ConnectException::class.java)) {
log.debug("Failed to connect to upstream: ${err.message}")
} else {
log.warn("Failed to validate upstream", err)
}
} }
.single()
.onErrorReturn(UpstreamAvailability.UNAVAILABLE) .onErrorReturn(UpstreamAvailability.UNAVAILABLE)
} }
fun start(): Flux<UpstreamAvailability> { fun start(): Flux<UpstreamAvailability> {
return Flux.interval(Duration.ofSeconds(15)) return Flux.interval(Duration.ofSeconds(15))
.subscribeOn(scheduler)
.flatMap { .flatMap {
validate() validate()
} }