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

fixed head subscription recovery

Browse Source
This commit is contained in:
Maxksim Fomenkov
2022-10-06 10:57:24 +03:00
parent d8efbcff47
commit 8c0d3faadd
2 changed files with 4 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
View File

@@ -104,6 +104,8 @@ class GrpcHead(
blocks = blocks.onErrorContinue { err, _ -> blocks = blocks.onErrorContinue { err, _ ->
log.error("Head subscription error. ${err.javaClass.name}:${err.message}", err) log.error("Head subscription error. ${err.javaClass.name}:${err.message}", err)
}.doOnNext{
log.info("Received block ${it.height}")
} }
headSubscription = super.follow(blocks) headSubscription = super.follow(blocks)

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -44,7 +44,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import reactor.core.Disposable import reactor.core.Disposable
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import java.net.ConnectException import java.io.IOException
import java.time.Duration import java.time.Duration
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantLock
@@ -92,7 +92,7 @@ class GrpcUpstreams(
.flatMap { .flatMap {
client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
}.onErrorContinue { t, _ -> }.onErrorContinue { t, _ ->
if (ExceptionUtils.indexOfType(t, ConnectException::class.java) >= 0) { if (ExceptionUtils.indexOfType(t, IOException::class.java) >= 0) {
log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})") log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})")
known.values.forEach { known.values.forEach {
it.setStatus(UpstreamAvailability.UNAVAILABLE) it.setStatus(UpstreamAvailability.UNAVAILABLE)