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

add handler for not able to emit removal event

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

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

@@ -10,6 +10,7 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Service 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
@@ -26,6 +27,7 @@ class SubscribeNodeStatus(
companion object { companion object {
private val RETRY_TIMEOUT = Duration.ofSeconds(10) private val RETRY_TIMEOUT = Duration.ofSeconds(10)
private val log = LoggerFactory.getLogger(SubscribeNodeStatus::class.java)
} }
fun subscribe(req: Mono<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> = fun subscribe(req: Mono<SubscribeNodeStatusRequest>): Flux<NodeStatusResponse> =
@@ -61,7 +63,10 @@ class SubscribeNodeStatus(
val removals = Flux.merge(multistreams.all().map { ms -> val removals = Flux.merge(multistreams.all().map { ms ->
ms.subscribeRemovedUpstreams().mapNotNull { up -> ms.subscribeRemovedUpstreams().mapNotNull { up ->
knownUpstreams[up.getId()]?.let { knownUpstreams[up.getId()]?.let {
it.tryEmitNext(true) val result = it.tryEmitNext(true)
if (result.isFailure) {
log.warn("Unable to emit event about removal of an upstream - ${result.toString()}")
}
knownUpstreams.remove(up.getId()) knownUpstreams.remove(up.getId())
NodeStatusResponse.newBuilder() NodeStatusResponse.newBuilder()
.setNodeId(up.getId()) .setNodeId(up.getId())