From b769446b0ebd02bd6bc9fd9da3007ef1b7352273 Mon Sep 17 00:00:00 2001 From: Termina1 Date: Mon, 9 Jan 2023 20:19:37 +0200 Subject: [PATCH] add handler for not able to emit removal event --- .../io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index a526f0c1..f3652c0d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -10,6 +10,7 @@ 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.slf4j.LoggerFactory import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -26,6 +27,7 @@ class SubscribeNodeStatus( companion object { private val RETRY_TIMEOUT = Duration.ofSeconds(10) + private val log = LoggerFactory.getLogger(SubscribeNodeStatus::class.java) } fun subscribe(req: Mono): Flux = @@ -61,7 +63,10 @@ class SubscribeNodeStatus( val removals = Flux.merge(multistreams.all().map { ms -> ms.subscribeRemovedUpstreams().mapNotNull { up -> 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()) NodeStatusResponse.newBuilder() .setNodeId(up.getId())