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

subscribe status send current upstream state on subscription

Browse Source
This commit is contained in:
Termina1
2023-01-12 13:05:47 +02:00
parent 62a581a378
commit f298b99251
1 changed files with 7 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,20 +29,6 @@ class SubscribeNodeStatus(
fun subscribe(req: Mono<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> = fun subscribe(req: Mono<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> =
req.flatMapMany { req.flatMapMany {
val knownUpstreams = ConcurrentHashMap<String, Sinks.Many<Boolean>>() val knownUpstreams = ConcurrentHashMap<String, Sinks.Many<Boolean>>()
// send known upstreams details immediately
val descriptions = Flux.fromIterable(
multistreams.all()
.flatMap { multiStream ->
multiStream.getAll().map { up ->
NodeStatusResponse.newBuilder()
.setNodeId(up.getId())
.setDescription(buildDescription(multiStream.chain, up))
.setStatus(buildStatus(up.getStatus(), up.getHead().getCurrentHeight()))
.build()
}
}
)
// subscribe on head/status updates for known upstreams // subscribe on head/status updates for known upstreams
val upstreamUpdates = Flux.merge( val upstreamUpdates = Flux.merge(
multistreams.all() multistreams.all()
@@ -102,7 +88,7 @@ class SubscribeNodeStatus(
} }
) )
Flux.concat(descriptions, Flux.merge(upstreamUpdates, multiStreamUpdates, removals)) Flux.merge(upstreamUpdates, multiStreamUpdates, removals)
} }
private fun subscribeUpstreamUpdates( private fun subscribeUpstreamUpdates(
@@ -120,8 +106,12 @@ class SubscribeNodeStatus(
.setDescription(buildDescription(chain, upstream)) .setDescription(buildDescription(chain, upstream))
.build() .build()
} }
val currentState = Mono.just(NodeStatusResponse.newBuilder()
return statuses .setNodeId(upstream.getId())
.setDescription(buildDescription(chain, 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(chain: Chain, up: Upstream): NodeDescription.Builder =