fix subscribe node status for available upstream subscriptions (#276)

This commit is contained in:
Vyacheslav
2023-08-09 12:37:04 +03:00
committed by GitHub
parent cea58efd9b
commit dfbdb0618f
8 changed files with 42 additions and 5 deletions

View File

@@ -153,7 +153,7 @@ class SubscribeNodeStatus(
.build()
}
)
.addAllSupportedSubscriptions(ms.getEgressSubscription().getAvailableTopics())
.addAllSupportedSubscriptions(up.getSubscriptionTopics())
.addAllSupportedMethods(up.getMethods().getSupportedMethods())
(up as? GrpcUpstream)?.let {
it.getBuildInfo().version?.let { version ->

View File

@@ -107,6 +107,10 @@ abstract class Multistream(
}
}
override fun getSubscriptionTopics(): List<String> {
return getEgressSubscription().getAvailableTopics()
}
private fun removeUpstreamMeters(upstreamId: String) {
meters[upstreamId]?.forEach {
Metrics.globalRegistry.remove(it)

View File

@@ -38,6 +38,7 @@ interface Upstream {
fun getLag(): Long?
fun getLabels(): Collection<UpstreamsConfig.Labels>
fun getMethods(): CallMethods
fun getSubscriptionTopics(): List<String>
fun getId(): String
fun getCapabilities(): Set<Capability>
fun isGrpc(): Boolean

View File

@@ -57,6 +57,10 @@ open class BitcoinRpcUpstream(
return directApi
}
override fun getSubscriptionTopics(): List<String> {
return listOf()
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return listOf(UpstreamsConfig.Labels())
}

View File

@@ -100,6 +100,15 @@ open class EthereumLikeRpcUpstream(
return connector.getIngressSubscription()
}
override fun getSubscriptionTopics(): List<String> {
val subs = if (getCapabilities().contains(Capability.WS_HEAD)) {
listOf(EthereumEgressSubscription.METHOD_NEW_HEADS, EthereumEgressSubscription.METHOD_LOGS)
} else {
listOf()
}
return getIngressSubscription().getAvailableTopics().plus(subs).toSet().toList()
}
override fun getHead(): Head {
return connector.getHead()
}

View File

@@ -84,6 +84,11 @@ class BitcoinGrpcUpstream(
)
block
}
override fun getSubscriptionTopics(): List<String> {
return listOf()
}
private val reloadBlock: Function<BlockContainer, Publisher<BlockContainer>> = Function { existingBlock ->
// head comes without transaction data
// need to download transactions for the block

View File

@@ -91,6 +91,10 @@ open class EthereumGrpcUpstream(
block
}
override fun getSubscriptionTopics(): List<String> {
return subscriptionTopics
}
private val reloadBlock: Function<BlockContainer, Publisher<BlockContainer>> = Function { existingBlock ->
// head comes without transaction data
// need to download transactions for the block
@@ -118,6 +122,7 @@ open class EthereumGrpcUpstream(
)
private var capabilities: Set<Capability> = emptySet()
private val buildInfo: BuildInfo = BuildInfo()
private var subscriptionTopics = listOf<String>()
private val defaultReader: JsonRpcReader = client.getReader()
var timeout = Defaults.timeout
@@ -152,7 +157,10 @@ open class EthereumGrpcUpstream(
capabilities = newCapabilities
}
conf.status?.let { status -> onStatus(status, upstreamStatusChanged) }
return buildInfoChanged || upstreamStatusChanged
val subsChanged = (conf.supportedSubscriptionsList != subscriptionTopics).also {
subscriptionTopics = conf.supportedSubscriptionsList
}
return buildInfoChanged || upstreamStatusChanged || subsChanged
}
override fun getQuorumByLabel(): QuorumForLabels {

View File

@@ -19,7 +19,6 @@ package io.emeraldpay.dshackle.upstream.grpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.config.ChainsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
@@ -94,8 +93,8 @@ open class EthereumPosGrpcUpstream(
private val buildInfo: BuildInfo = BuildInfo()
private val defaultReader: JsonRpcReader = client.getReader()
private val timeout = Defaults.timeout
private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
private var subscriptionTopics = listOf<String>()
override fun start() {
}
@@ -107,6 +106,10 @@ open class EthereumPosGrpcUpstream(
override fun stop() {
}
override fun getSubscriptionTopics(): List<String> {
return subscriptionTopics
}
override fun getBuildInfo(): BuildInfo {
return buildInfo
}
@@ -119,7 +122,10 @@ open class EthereumPosGrpcUpstream(
capabilities = newCapabilities
}
conf.status?.let { status -> onStatus(status, upstreamStatusChanged) }
return buildInfoChanged || upstreamStatusChanged
val subsChanged = (conf.supportedSubscriptionsList != subscriptionTopics).also {
subscriptionTopics = conf.supportedSubscriptionsList
}
return buildInfoChanged || upstreamStatusChanged || subsChanged
}
override fun getQuorumByLabel(): QuorumForLabels {