handle subscription updates for upstream removals separately

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

@@ -30,7 +30,7 @@ class SubscribeNodeStatus(
 
     fun subscribe(req: Mono<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> =
         req.flatMapMany { request ->
-            val knownUpstreams = ConcurrentHashMap<String, Boolean>()
+            val knownUpstreams = ConcurrentHashMap<String, Sinks.Many<Boolean>>()
             val duration = Duration.ofMillis(request.timespan)
             // send known upstreams details immediately
             val descriptions = Flux.fromIterable(
@@ -51,12 +51,27 @@ class SubscribeNodeStatus(
                 multistreams.all()
                     .flatMap { ms ->
                         ms.getAll().map { up ->
-                            knownUpstreams[up.getId()] = true
+                            knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
-                            subscribeUpstreamUpdates(ms.chain, up, duration) { r -> knownUpstreams.remove(r) }
+                            subscribeUpstreamUpdates(ms.chain, up, duration, knownUpstreams[up.getId()]!!)
                         }
                     }
             )
 
+            // stop removed upstreams update fluxes
+            val removals = Flux.merge(multistreams.all().map { ms ->
+                ms.subscribeRemovedUpstreams().mapNotNull { up ->
+                    knownUpstreams[up.getId()]?.let {
+                        it.tryEmitNext(true)
+                        knownUpstreams.remove(up.getId())
+                        NodeStatusResponse.newBuilder()
+                            .setNodeId(up.getId())
+                            .setDescription(buildDescription(ms.chain, up))
+                            .setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight()))
+                            .build()
+                    }
+                }
+            })
+
             // subscribe on head/status updates for just added upstreams
             val multiStreamUpdates = Flux.merge(
                 multistreams.all()
@@ -66,10 +81,10 @@ class SubscribeNodeStatus(
                                 it.getId()
                             }
                             .filter {
-                                !knownUpstreams.getOrDefault(it.getId(), false)
+                                !knownUpstreams.contains(it.getId())
                             }
                             .flatMap {
-                                knownUpstreams[it.getId()] = true
+                                knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
                                 Flux.concat(
                                     Mono.just(
                                         NodeStatusResponse.newBuilder()
@@ -78,24 +93,22 @@ class SubscribeNodeStatus(
                                             .setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight()))
                                             .build()
                                     ),
-                                    subscribeUpstreamUpdates(ms.chain, it, duration) { r -> knownUpstreams.remove(r) }
+                                    subscribeUpstreamUpdates(ms.chain, it, duration, knownUpstreams[it.getId()]!!)
                                 )
                             }
                     }
             )
 
-            Flux.concat(descriptions, Flux.merge(upstreamUpdates, multiStreamUpdates))
+            Flux.concat(descriptions, Flux.merge(upstreamUpdates, multiStreamUpdates, removals))
         }
 
     private fun subscribeUpstreamUpdates(
         chain: Chain,
         upstream: Upstream,
         timespan: Duration,
-        onUnavailable: Consumer<String>
+        cancel: Sinks.Many<Boolean>
     ): Flux<NodeStatusResponse> {
         val retry = Sinks.many().multicast().directBestEffort<Boolean>()
-        val cancel = Sinks.many().multicast().directBestEffort<Boolean>()
-
         val heads = Mono.just(upstream)
             .repeatWhen {
                 retry.asFlux()
@@ -119,15 +132,8 @@ class SubscribeNodeStatus(
 
         val statuses = upstream.observeStatus()
             .distinctUntilChanged()
-            .takeUntil {
+            .takeUntilOther(cancel.asFlux())
-                it == UpstreamAvailability.UNAVAILABLE && !upstream.isGrpc()
-            }
             .map {
-                if (it == UpstreamAvailability.UNAVAILABLE && !upstream.isGrpc()) {
-                    onUnavailable.accept(upstream.getId())
-                    // cancel head subscription & reconnections when upstream becomes unavailable
-                    cancel.tryEmitNext(true)
-                }
                 NodeStatusResponse.newBuilder()
                     .setNodeId(upstream.getId())
                     .setStatus(buildStatus(it, upstream.getHead().getCurrentHeight()))

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

@@ -78,6 +78,9 @@ abstract class Multistream(
     private val addedUpstreams = Sinks.many()
         .multicast()
         .directBestEffort<Upstream>()
+    private val removedUpstreams = Sinks.many()
+        .multicast()
+        .directBestEffort<Upstream>()
 
     init {
         UpstreamAvailability.values().forEach { status ->
@@ -342,6 +345,7 @@ abstract class Multistream(
             eventLock.withLock {
                 if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) {
                     removeUpstream(event.upstream.getId()).takeIf { it }?.let {
+                        removedUpstreams.tryEmitNext(event.upstream)
                         log.warn("Upstream ${event.upstream.getId()} with chain $chain has been removed")
                     }
                 } else {
@@ -370,6 +374,9 @@ abstract class Multistream(
     fun subscribeAddedUpstreams(): Flux<Upstream> =
         addedUpstreams.asFlux()
 
+    fun subscribeRemovedUpstreams(): Flux<Upstream> =
+        removedUpstreams.asFlux()
+
     // --------------------------------------------------------------------------------------------------------
 
     class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())