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

Merge pull request #119 from p2p-org/add_supportedSubs

Browse Source 
Add supported subscription methods to SubscribeNodeStatus response
This commit is contained in:
KirillPamPam
2023-01-23 19:44:48 +04:00
committed by GitHub
parent 78a27c9782 0f7fd35b12
commit 731ee2c747
2 changed files with 12 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
emerald-grpc

Submodule emerald-grpc updated: f5543c9603...fddf154c34

21
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt
View File

@@ -6,8 +6,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass.NodeStatus
import io.emeraldpay.api.proto.BlockchainOuterClass.NodeStatusResponse
import io.emeraldpay.api.proto.BlockchainOuterClass.SubscribeNodeStatusRequest
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import org.slf4j.LoggerFactory
@@ -35,7 +35,7 @@ class SubscribeNodeStatus(
.flatMap { ms ->
ms.getAll().map { up ->
knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
subscribeUpstreamUpdates(ms.chain, up, knownUpstreams[up.getId()]!!)
subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
}
}
)
@@ -53,7 +53,7 @@ class SubscribeNodeStatus(
knownUpstreams.remove(up.getId())
NodeStatusResponse.newBuilder()
.setNodeId(up.getId())
.setDescription(buildDescription(ms.chain, up))
.setDescription(buildDescription(ms, up))
.setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight()))
.build()
}
@@ -78,11 +78,11 @@ class SubscribeNodeStatus(
Mono.just(
NodeStatusResponse.newBuilder()
.setNodeId(it.getId())
.setDescription(buildDescription(ms.chain, it))
.setDescription(buildDescription(ms, it))
.setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight()))
.build()
),
subscribeUpstreamUpdates(ms.chain, it, knownUpstreams[it.getId()]!!)
subscribeUpstreamUpdates(ms, it, knownUpstreams[it.getId()]!!)
)
}
}
@@ -92,7 +92,7 @@ class SubscribeNodeStatus(
}
private fun subscribeUpstreamUpdates(
chain: Chain,
ms: Multistream,
upstream: Upstream,
cancel: Sinks.Many<Boolean>
): Flux<NodeStatusResponse> {
@@ -103,22 +103,22 @@ class SubscribeNodeStatus(
NodeStatusResponse.newBuilder()
.setNodeId(upstream.getId())
.setStatus(buildStatus(it, upstream.getHead().getCurrentHeight()))
.setDescription(buildDescription(chain, upstream))
.setDescription(buildDescription(ms, upstream))
.build()
}
val currentState = Mono.just(
NodeStatusResponse.newBuilder()
.setNodeId(upstream.getId())
.setDescription(buildDescription(chain, upstream))
.setDescription(buildDescription(ms, upstream))
.setStatus(buildStatus(upstream.getStatus(), upstream.getHead().getCurrentHeight()))
.build()
)
return Flux.concat(currentState, statuses)
}
private fun buildDescription(chain: Chain, up: Upstream): NodeDescription.Builder =
private fun buildDescription(ms: Multistream, up: Upstream): NodeDescription.Builder =
NodeDescription.newBuilder()
.setChain(Common.ChainRef.forNumber(chain.id))
.setChain(Common.ChainRef.forNumber(ms.chain.id))
.setNodeId(up.nodeId().toInt())
.addAllNodeLabels(
up.getLabels().map { nodeLabels ->
@@ -134,6 +134,7 @@ class SubscribeNodeStatus(
.build()
}
)
.addAllSupportedSubscriptions(ms.getEgressSubscription().getAvailableTopics())
.addAllSupportedMethods(up.getMethods().getSupportedMethods())
private fun buildStatus(status: UpstreamAvailability, height: Long?): NodeStatus.Builder =