Send head updates (#236)

This commit is contained in:
KirillPamPam
2023-06-28 13:34:13 +04:00
committed by GitHub
parent ab9dd6de5a
commit 3e2cae4793

View File

@@ -16,6 +16,7 @@ import org.springframework.stereotype.Service
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import reactor.core.publisher.Sinks import reactor.core.publisher.Sinks
import java.time.Duration
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
@Service @Service
@@ -100,6 +101,21 @@ class SubscribeNodeStatus(
upstream: Upstream, upstream: Upstream,
cancel: Sinks.Many<Boolean> cancel: Sinks.Many<Boolean>
): Flux<NodeStatusResponse> { ): Flux<NodeStatusResponse> {
val heads = upstream.getHead().getFlux()
.takeUntilOther(cancel.asFlux())
.sample(Duration.ofSeconds(2))
.map { block ->
NodeStatusResponse.newBuilder()
.setDescription(
NodeDescription.newBuilder()
.setNodeId(upstream.nodeId().toInt())
.setChain(Common.ChainRef.forNumber(ms.chain.id))
.build()
)
.setNodeId(upstream.getId())
.setStatus(buildStatus(upstream.getStatus(), block.height))
.build()
}
val statuses = upstream.observeStatus() val statuses = upstream.observeStatus()
.takeUntilOther(cancel.asFlux()) .takeUntilOther(cancel.asFlux())
.map { .map {
@@ -116,7 +132,7 @@ class SubscribeNodeStatus(
.setStatus(buildStatus(upstream.getStatus(), upstream.getHead().getCurrentHeight())) .setStatus(buildStatus(upstream.getStatus(), upstream.getHead().getCurrentHeight()))
.build() .build()
) )
return Flux.concat(currentState, statuses) return Flux.concat(currentState, Flux.merge(statuses, heads))
} }
private fun buildDescription(ms: Multistream, up: Upstream): NodeDescription.Builder { private fun buildDescription(ms: Multistream, up: Upstream): NodeDescription.Builder {