From dfbdb0618fa8b2a00a08a30e3ed2f313a3b354ae Mon Sep 17 00:00:00 2001 From: Vyacheslav Date: Wed, 9 Aug 2023 12:37:04 +0300 Subject: [PATCH] fix subscribe node status for available upstream subscriptions (#276) --- .../emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt | 2 +- .../io/emeraldpay/dshackle/upstream/Multistream.kt | 4 ++++ .../io/emeraldpay/dshackle/upstream/Upstream.kt | 1 + .../dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt | 4 ++++ .../upstream/ethereum/EthereumLikeRpcUpstream.kt | 9 +++++++++ .../dshackle/upstream/grpc/BitcoinGrpcUpstream.kt | 5 +++++ .../dshackle/upstream/grpc/EthereumGrpcUpstream.kt | 10 +++++++++- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 12 +++++++++--- 8 files changed, 42 insertions(+), 5 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index b25ca37c..e2b5dd4d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -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 -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index d436c57e..e20afede 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -107,6 +107,10 @@ abstract class Multistream( } } + override fun getSubscriptionTopics(): List { + return getEgressSubscription().getAvailableTopics() + } + private fun removeUpstreamMeters(upstreamId: String) { meters[upstreamId]?.forEach { Metrics.globalRegistry.remove(it) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index ab66781b..d1d1b884 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -38,6 +38,7 @@ interface Upstream { fun getLag(): Long? fun getLabels(): Collection fun getMethods(): CallMethods + fun getSubscriptionTopics(): List fun getId(): String fun getCapabilities(): Set fun isGrpc(): Boolean diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt index cb6e4f6e..f1bb1495 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -57,6 +57,10 @@ open class BitcoinRpcUpstream( return directApi } + override fun getSubscriptionTopics(): List { + return listOf() + } + override fun getLabels(): Collection { return listOf(UpstreamsConfig.Labels()) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt index 025c40d5..e4e12c8c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt @@ -100,6 +100,15 @@ open class EthereumLikeRpcUpstream( return connector.getIngressSubscription() } + override fun getSubscriptionTopics(): List { + 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() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 8bbb6424..9423f837 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -84,6 +84,11 @@ class BitcoinGrpcUpstream( ) block } + + override fun getSubscriptionTopics(): List { + return listOf() + } + private val reloadBlock: Function> = Function { existingBlock -> // head comes without transaction data // need to download transactions for the block diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 404139bc..75a80675 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -91,6 +91,10 @@ open class EthereumGrpcUpstream( block } + override fun getSubscriptionTopics(): List { + return subscriptionTopics + } + private val reloadBlock: Function> = 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 = emptySet() private val buildInfo: BuildInfo = BuildInfo() + private var subscriptionTopics = listOf() 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 11a958fc..3692a4e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -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() override fun start() { } @@ -107,6 +106,10 @@ open class EthereumPosGrpcUpstream( override fun stop() { } + override fun getSubscriptionTopics(): List { + 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 {