Fix remove upstream, dispose lower sub (#373)

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<Pair<String, Chain>>,
         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()
             }
         }
     }

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),