subscribe node status

src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt

@@ -47,6 +47,7 @@ class BlockchainRpc(
     private val describe: Describe,
     private val subscribeStatus: SubscribeStatus,
     private val estimateFee: EstimateFee,
+    private val subscribeNodeStatus: SubscribeNodeStatus,
     @Qualifier("rpcScheduler")
     private val scheduler: Scheduler
 ) : ReactorBlockchainGrpc.BlockchainImplBase() {
@@ -220,6 +221,10 @@ class BlockchainRpc(
             .doOnError { failMetric.increment() }
     }
 
+    override fun subscribeNodeStatus(request: Mono<BlockchainOuterClass.SubscribeNodeStatusRequest>): Flux<BlockchainOuterClass.NodeStatusResponse> {
+        return subscribeNodeStatus.subscribe(request).subscribeOn(scheduler).doOnError { failMetric.increment() }
+    }
+
     class RequestMetrics(val chain: Chain) {
         val nativeCallMetric = Counter.builder("request.grpc.request")
             .tag("type", "nativeCall")

src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt

@@ -0,0 +1,113 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.BlockchainOuterClass.NodeDescription
+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.Upstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.SignalType
+import reactor.core.publisher.Sinks
+import java.time.Duration
+import java.util.concurrent.ConcurrentHashMap
+
+@Service
+class SubscribeNodeStatus(
+    private val multistreams: CurrentMultistreamHolder
+) {
+
+    fun subscribe(req: Mono<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> =
+        req.flatMapMany {
+            val knownUpstreams = ConcurrentHashMap<String, 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
+            val upstreamUpdates = Flux.merge(multistreams.all().flatMap { ms ->
+                ms.getAll().map { up ->
+                    knownUpstreams[up.getId()] = true
+                    subscribeUpstreamUpdates(ms.chain, up)
+                }
+            })
+
+            //subscribe on head/status updates for just added upstreams
+            val muliStreamUpdates = Flux.merge(multistreams.all().map { ms ->
+                ms.subscribeAddedUpstreams()
+                    .distinctUntilChanged {
+                        it.getId()
+                    }
+                    .filter { knownUpstreams[it.getId()] != true }.flatMap {
+                        knownUpstreams[it.getId()] = true
+                        Flux.concat(
+                            Mono.just(
+                                NodeStatusResponse.newBuilder()
+                                    .setNodeId(it.getId())
+                                    .setDescription(buildDescription(ms.chain, it))
+                                    .setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight()))
+                                    .build()
+                            ), subscribeUpstreamUpdates(ms.chain, it)
+                        )
+                    }
+            })
+
+            Flux.concat(descriptions, Flux.merge(upstreamUpdates, muliStreamUpdates))
+        }
+
+    private fun subscribeUpstreamUpdates(chain: Chain, upstream: Upstream): Flux<NodeStatusResponse> {
+        val r = Sinks.many().multicast().directBestEffort<Boolean>()
+        val heads = Mono.just(upstream).repeatWhen { r.asFlux() }.sample(Duration.ofSeconds(10)).flatMap { up ->
+            up.getHead().getFlux().map { block ->
+                NodeStatusResponse.newBuilder()
+                    .setNodeId(up.getId())
+                    .setStatus(buildStatus(up.getStatus(), block.height))
+                    .build()
+            }.doFinally {
+                if (it == SignalType.ON_COMPLETE) {
+                    r.tryEmitNext(true)
+                }
+            }
+        }
+
+        val statuses = upstream.observeStatus().map {
+            NodeStatusResponse.newBuilder()
+                .setNodeId(upstream.getId())
+                .setStatus(buildStatus(it, upstream.getHead().getCurrentHeight()))
+                .setDescription(buildDescription(chain, upstream))
+                .build()
+        }
+        return Flux.merge(heads.distinctUntilChanged(), statuses.distinctUntilChanged())
+    }
+
+    private fun buildDescription(chain: Chain, up: Upstream): NodeDescription.Builder =
+        NodeDescription.newBuilder()
+            .setChain(Common.ChainRef.forNumber(chain.id))
+            .addAllLabels(up.getLabels().flatMap { labels ->
+                labels.map {
+                    BlockchainOuterClass.Label.newBuilder()
+                        .setName(it.key)
+                        .setValue(it.value)
+                        .build()
+                }
+            })
+            .addAllSupportedMethods(up.getMethods().getSupportedMethods())
+
+    private fun buildStatus(status: UpstreamAvailability, height: Long?): NodeStatus.Builder =
+        NodeStatus.newBuilder()
+            .setAvailability(BlockchainOuterClass.AvailabilityEnum.forNumber(status.grpcId))
+            .setCurrentHeight(height ?: 0)
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt

@@ -45,6 +45,9 @@ open class CurrentMultistreamHolder(
         return chainMapping.getValue(chain).isAvailable()
     }
 
+    override fun all(): List<Multistream> =
+        chainMapping.values.toList()
+
     @PreDestroy
     fun shutdown() {
         log.info("Closing upstream connections...")

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -39,6 +39,7 @@ import org.springframework.core.annotation.Order
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
 import java.time.Duration
 import java.time.Instant
 import java.util.Locale
@@ -76,6 +77,9 @@ abstract class Multistream(
     private var capabilities: Set<Capability> = emptySet()
     private val removed: MutableMap<String, Upstream> = HashMap()
     private val meters: MutableMap<String, Meter.Id> = HashMap()
+    private val addedUpstreams = Sinks.many()
+        .multicast()
+        .directBestEffort<Upstream>()
 
     init {
         UpstreamAvailability.values().forEach { status ->
@@ -350,6 +354,7 @@ abstract class Multistream(
                         if (!started) {
                             start()
                         }
+                        addedUpstreams.tryEmitNext(event.upstream)
                         log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
                     }
                 }
@@ -364,6 +369,9 @@ abstract class Multistream(
         return upstreams.any { matcher.matches(it) }
     }
 
+    fun subscribeAddedUpstreams(): Flux<Upstream> =
+        addedUpstreams.asFlux()
+
     // --------------------------------------------------------------------------------------------------------
 
     class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())

src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt

@@ -25,4 +25,6 @@ interface MultistreamHolder {
     fun getUpstream(chain: Chain): Multistream
     fun getAvailable(): List<Chain>
     fun isAvailable(chain: Chain): Boolean
+
+    fun all(): List<Multistream>
 }