diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt index c9b95c41..81b3f3b7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt @@ -2,11 +2,13 @@ package io.emeraldpay.dshackle.config.reload import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.Global.Companion.chainById import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.startup.ConfiguredUpstreams import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import org.springframework.stereotype.Component +import java.util.stream.Collectors @Component open class ReloadConfigUpstreamService( @@ -20,19 +22,21 @@ open class ReloadConfigUpstreamService( upstreamsToAdd: Set>, newUpstreamsConfig: UpstreamsConfig, ) { + val newUpstreamsCount = newUpstreamsConfig.upstreams.stream() + .collect( + Collectors.groupingBy( + { chainById(it.chain) }, + Collectors.counting(), + ), + ) + val usedChains = removeUpstreams(chainsToReload, upstreamsToRemove) addUpstreams(newUpstreamsConfig, chainsToReload, upstreamsToAdd.map { it.first }.toSet()) usedChains.forEach { chain -> - val upstreamsToRemovePerChain = upstreamsToRemove.filter { it.second == chain } - val upstreamsToAddPerChain = upstreamsToAdd.filter { it.second == chain } - - if (upstreamsToAddPerChain.isEmpty() && upstreamsToRemovePerChain.isNotEmpty()) { - multistreamHolder.getUpstream(chain) - .run { - this.stop() - } + if (newUpstreamsCount[chain] == null) { + multistreamHolder.getUpstream(chain).stop() } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index 56f87c86..997aafce 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt @@ -48,6 +48,7 @@ open class GenericUpstream( private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig) private var validatorSubscription: Disposable? = null private var validationSettingsSubscription: Disposable? = null + private var lowerBlockDetectorSubscription: Disposable? = null private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false) protected val connector: GenericConnector = connectorFactory.create(this, chain) @@ -180,6 +181,8 @@ open class GenericUpstream( validatorSubscription = null livenessSubscription?.dispose() livenessSubscription = null + lowerBlockDetectorSubscription?.dispose() + lowerBlockDetectorSubscription = null disposeValidationSettingsSubscription() connector.stop() } @@ -197,7 +200,7 @@ open class GenericUpstream( } private fun detectLowerBlock() { - lowerBoundBlockDetector.lowerBlock() + lowerBlockDetectorSubscription = lowerBoundBlockDetector.lowerBlock() .subscribe { stateEventStream.emitNext( UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED),