Now grpchead have correct name (#175)

This commit is contained in:
a10zn8
2023-03-19 18:59:03 +03:00
committed by GitHub
parent d281cd8857
commit b1f302aa60
7 changed files with 15 additions and 41 deletions

View File

@@ -123,10 +123,6 @@ abstract class DefaultUpstream(
return this.status.get().lag return this.status.get().lag
} }
override fun getId(): String {
return id
}
override fun getOptions(): UpstreamsConfig.Options { override fun getOptions(): UpstreamsConfig.Options {
return options return options
} }
@@ -148,5 +144,9 @@ abstract class DefaultUpstream(
return quorumByLabel return quorumByLabel
} }
override fun getId(): String {
return id
}
data class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability) data class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability)
} }

View File

@@ -32,7 +32,7 @@ abstract class BitcoinUpstream(
callMethods: CallMethods, callMethods: CallMethods,
node: QuorumForLabels.QuorumItem, node: QuorumForLabels.QuorumItem,
val esploraClient: EsploraClient? = null, val esploraClient: EsploraClient? = null,
private val chainConfig: ChainsConfig.ChainConfig chainConfig: ChainsConfig.ChainConfig
) : DefaultUpstream(id, 0.toByte(), options, role, callMethods, node, chainConfig) { ) : DefaultUpstream(id, 0.toByte(), options, role, callMethods, node, chainConfig) {
constructor( constructor(

View File

@@ -100,7 +100,7 @@ class BitcoinGrpcUpstream(
} }
} }
private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
private val timeout = Defaults.timeout private val timeout = Defaults.timeout
private var capabilities: Set<Capability> = emptySet() private var capabilities: Set<Capability> = emptySet()

View File

@@ -57,7 +57,7 @@ open class EthereumGrpcUpstream(
role: UpstreamsConfig.UpstreamRole, role: UpstreamsConfig.UpstreamRole,
private val chain: Chain, private val chain: Chain,
private val remote: ReactorBlockchainStub, private val remote: ReactorBlockchainStub,
private val client: JsonRpcGrpcClient, client: JsonRpcGrpcClient,
overrideLabels: UpstreamsConfig.Labels?, overrideLabels: UpstreamsConfig.Labels?,
chainConfig: ChainsConfig.ChainConfig chainConfig: ChainsConfig.ChainConfig
) : EthereumUpstream( ) : EthereumUpstream(
@@ -107,7 +107,7 @@ open class EthereumGrpcUpstream(
private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
private var capabilities: Set<Capability> = emptySet() private var capabilities: Set<Capability> = emptySet()
private val defaultReader: JsonRpcReader = client.getReader() private val defaultReader: JsonRpcReader = client.getReader()

View File

@@ -30,32 +30,24 @@ import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.etherjar.rpc.RpcException
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.math.BigInteger import java.math.BigInteger
import java.time.Instant import java.time.Instant
import java.util.Locale import java.util.Locale
import java.util.concurrent.TimeoutException
import java.util.function.Function import java.util.function.Function
open class EthereumPosGrpcUpstream( open class EthereumPosGrpcUpstream(
private val parentId: String, parentId: String,
hash: Byte, hash: Byte,
role: UpstreamsConfig.UpstreamRole, role: UpstreamsConfig.UpstreamRole,
private val chain: Chain, chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
client: JsonRpcGrpcClient, client: JsonRpcGrpcClient,
nodeRating: Int, nodeRating: Int,
@@ -86,29 +78,8 @@ open class EthereumPosGrpcUpstream(
block block
} }
private val reloadBlock: Function<BlockContainer, Publisher<BlockContainer>> = Function { existingBlock ->
// head comes without transaction data
// need to download transactions for the block
defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(existingBlock.hash.toHexWithPrefix(), false)))
.flatMap(JsonRpcResponse::requireResult)
.map {
BlockContainer.fromEthereumJson(it, getId())
}
.timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))
.doOnError { t ->
setStatus(UpstreamAvailability.UNAVAILABLE)
val msg = "Failed to download block data for chain $chain on $parentId"
if (t is RpcException || t is TimeoutException) {
log.warn("$msg. Message: ${t.message}")
} else {
log.error(msg, t)
}
}
}
private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, null, NoChoiceWithPriorityForkChoice(nodeRating, parentId)) private val grpcHead = GrpcHead(getId(), chain, this, remote, blockConverter, null, NoChoiceWithPriorityForkChoice(nodeRating, parentId))
private var capabilities: Set<Capability> = emptySet() private var capabilities: Set<Capability> = emptySet()
private val defaultReader: JsonRpcReader = client.getReader() private val defaultReader: JsonRpcReader = client.getReader()

View File

@@ -35,6 +35,7 @@ import java.time.Duration
import java.util.function.Function import java.util.function.Function
class GrpcHead( class GrpcHead(
id: String,
private val chain: Chain, private val chain: Chain,
private val parent: DefaultUpstream, private val parent: DefaultUpstream,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
@@ -47,7 +48,7 @@ class GrpcHead(
*/ */
private val enhancer: Function<BlockContainer, Publisher<BlockContainer>>?, private val enhancer: Function<BlockContainer, Publisher<BlockContainer>>?,
private val forkChoice: ForkChoice private val forkChoice: ForkChoice
) : AbstractHead(forkChoice), Lifecycle { ) : AbstractHead(forkChoice, upstreamId = id), Lifecycle {
companion object { companion object {
private val log = LoggerFactory.getLogger(GrpcHead::class.java) private val log = LoggerFactory.getLogger(GrpcHead::class.java)

View File

@@ -58,6 +58,7 @@ class GrpcHeadSpec extends Specification {
TestingCommons.blockForBitcoin(head.height) TestingCommons.blockForBitcoin(head.height)
} }
def head = new GrpcHead( def head = new GrpcHead(
"test",
Chain.BITCOIN, Chain.BITCOIN,
Stub(DefaultUpstream), Stub(DefaultUpstream),
client, client,
@@ -119,6 +120,7 @@ class GrpcHeadSpec extends Specification {
TestingCommons.blockForBitcoin(head.height) TestingCommons.blockForBitcoin(head.height)
} }
def head = new GrpcHead( def head = new GrpcHead(
"test",
Chain.BITCOIN, Chain.BITCOIN,
Stub(DefaultUpstream), Stub(DefaultUpstream),
client, client,