diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index 9967b156..93013e66 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -119,10 +119,11 @@ class SubscribeNodeStatus( val statuses = upstream.observeStatus() .distinctUntilChanged() - .takeUntil{it == UpstreamAvailability.UNAVAILABLE} + .takeUntil { it == UpstreamAvailability.UNAVAILABLE } .map { if (it == UpstreamAvailability.UNAVAILABLE) { onUnavailable.accept(upstream.getId()) + // cancel head subscription & reconnections when upstream becomes unavailable cancel.tryEmitNext(true) } NodeStatusResponse.newBuilder()