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

fix stylw

Browse Source
This commit is contained in:
Termina1
2023-01-09 20:38:15 +02:00
parent b769446b0e
commit ed9d39b733
1 changed files with 17 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,7 +18,6 @@ import reactor.core.publisher.SignalType
import reactor.core.publisher.Sinks
import java.time.Duration
import java.util.concurrent.ConcurrentHashMap
import java.util.function.Consumer
@Service
class SubscribeNodeStatus(
@@ -60,22 +59,25 @@ class SubscribeNodeStatus(
)
// stop removed upstreams update fluxes
val removals = Flux.merge(multistreams.all().map { ms ->
ms.subscribeRemovedUpstreams().mapNotNull { up ->
knownUpstreams[up.getId()]?.let {
val result = it.tryEmitNext(true)
if (result.isFailure) {
log.warn("Unable to emit event about removal of an upstream - ${result.toString()}")
val removals = Flux.merge(
multistreams.all()
.map { ms ->
ms.subscribeRemovedUpstreams().mapNotNull { up ->
knownUpstreams[up.getId()]?.let {
val result = it.tryEmitNext(true)
if (result.isFailure) {
log.warn("Unable to emit event about removal of an upstream - $result")
}
knownUpstreams.remove(up.getId())
NodeStatusResponse.newBuilder()
.setNodeId(up.getId())
.setDescription(buildDescription(ms.chain, up))
.setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight()))
.build()
}
}
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(