problem: starts subscription on init, may keep them forever

src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt

@@ -2,6 +2,8 @@ package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
 import java.io.Closeable
 import java.lang.IllegalStateException
 import java.time.Duration
@@ -10,26 +12,47 @@ class ChainUpstreams (
         val chain: Chain,
         private val upstreams: MutableList<Upstream>,
         targets: CallMethods
-) : AggregatedUpstream(targets) {
+) : AggregatedUpstream(targets), Lifecycle {
+
 
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0
     private var head: EthereumHead?
     private var lagObserver: HeadLagObserver? = null
+    private var subscription: Disposable? = null
 
     init {
         head = updateHead()
-        observeStatus()
+    }
+
+    override fun isRunning(): Boolean {
+        return subscription != null
+    }
+
+    override fun start() {
+        subscription = observeStatus()
                 .distinctUntilChanged()
                 .subscribe { printStatus() }
     }
 
-    internal fun updateHead(): EthereumHead {
+    override fun stop() {
-        val current = head
+        subscription?.dispose()
-        if (current != null && Closeable::class.java.isAssignableFrom(current.javaClass)) {
+        subscription = null
-            (current as Closeable).close()
+        head?.let {
+            if (it is Lifecycle) {
+                it.stop()
             }
-        lagObserver?.close()
+        }
+        lagObserver?.stop()
+    }
+
+    internal fun updateHead(): EthereumHead {
+        head?.let {
+            if (it is Lifecycle) {
+                it.stop()
+            }
+        }
+        lagObserver?.stop()
         lagObserver = null
         return if (upstreams.size == 1) {
             val upstream = upstreams.first()
@@ -37,6 +60,7 @@ class ChainUpstreams (
             upstream.getHead()
         } else {
             val newHead = EthereumHeadMerge(upstreams.map { it.getHead() })
+            newHead.start()
             val lagObserver = HeadLagObserver(newHead, upstreams)
             lagObserver.start()
             this.lagObserver = lagObserver

src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt

@@ -132,7 +132,9 @@ open class ConfiguredUpstreams(
         }
         if (rpcApi != null) {
             log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-            addUpstream(chain, EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain)))
+            val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain))
+            ethereumUpstream.start()
+            addUpstream(chain, ethereumUpstream)
         }
     }
 
@@ -165,6 +167,7 @@ open class ConfiguredUpstreams(
         if (current == null) {
             val created = ChainUpstreams(chain, ArrayList<Upstream>(), targetFor(chain))
             created.addUpstream(up)
+            created.start()
             chainMapping[chain] = created
             chainsBus.onNext(chain)
             return created

src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
@@ -10,13 +11,13 @@ import java.io.Closeable
 import java.util.concurrent.atomic.AtomicReference
 
 class EthereumHeadMerge(
-        private val upstreams: List<EthereumHead>
+        upstreams: List<EthereumHead>
-): EthereumHead, Closeable {
+): EthereumHead, Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java)
     private val flux: Flux<BlockJson<TransactionId>>
     private val head = AtomicReference<BlockJson<TransactionId>>(null)
-    private val subscription: Disposable
+    private var subscription: Disposable? = null
 
     init {
         val fluxes = upstreams.map { it.getFlux() }
@@ -30,13 +31,19 @@ class EthereumHeadMerge(
                 }
                 .publish()
                 .autoConnect()
+    }
 
+    override fun isRunning(): Boolean {
+        return subscription != null
+    }
 
+    override fun start() {
         subscription = Flux.from(flux).subscribe {
             head.set(it)
         }
     }
 
+
     override fun getHead(): Mono<BlockJson<TransactionId>> {
         val curr = head.get()
         if (curr != null) {
@@ -50,10 +57,8 @@ class EthereumHeadMerge(
                 .onBackpressureLatest()
     }
 
-    override fun close() {
+    override fun stop() {
-        if (!subscription.isDisposed) {
+        subscription?.dispose()
-            subscription.dispose()
-        }
     }
 
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt

@@ -5,6 +5,8 @@ import io.infinitape.etherjar.rpc.Batch
 import io.infinitape.etherjar.rpc.Commands
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.TopicProcessor
@@ -13,15 +15,16 @@ import java.util.concurrent.atomic.AtomicReference
 
 class EthereumRpcHead(
     private val api: EthereumApi
-): EthereumHead {
+): EthereumHead, Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java)
 
     private val head = AtomicReference<BlockJson<TransactionId>>(null)
     private val stream: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
+    private var refreshSubscription: Disposable? = null
 
-    fun start() {
+    override fun start() {
-        Flux.interval(Duration.ofSeconds(7))
+        refreshSubscription = Flux.interval(Duration.ofSeconds(7))
                 .flatMap {
                     val batch = Batch()
                     val f = batch.add(Commands.eth().blockNumber)
@@ -48,6 +51,16 @@ class EthereumRpcHead(
                 }
     }
 
+    override fun isRunning(): Boolean {
+        return refreshSubscription != null
+    }
+
+    override fun stop() {
+        refreshSubscription?.dispose()
+        refreshSubscription = null
+    }
+
+
     override fun getHead(): Mono<BlockJson<TransactionId>> {
         val current = head.get()
         if (current != null) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt

@@ -3,6 +3,9 @@ package io.emeraldpay.dshackle.upstream
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
+import java.io.Closeable
 
 open class EthereumUpstream(
         val chain: Chain,
@@ -11,7 +14,7 @@ open class EthereumUpstream(
         private val options: UpstreamsConfig.Options,
         val node: NodeDetailsList.NodeDetails,
         private val targets: CallMethods
-): DefaultUpstream() {
+): DefaultUpstream(), Lifecycle {
 
     constructor(chain: Chain, api: EthereumApi): this(chain, api, null,
             UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()),
@@ -24,21 +27,34 @@ open class EthereumUpstream(
     private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
 
     private val head: EthereumHead = this.createHead()
+    private var validatorSubscription: Disposable? = null
 
     init {
-        log.info("Configured for ${chain.chainName}")
         api.upstream = this
+    }
+
+    override fun start() {
+        log.info("Configured for ${chain.chainName}")
 
         if (options.disableValidation != null && options.disableValidation!!) {
             this.setLag(0)
             this.setStatus(UpstreamAvailability.OK)
         } else {
             val validator = UpstreamValidator(this, options)
-            validator.start()
+            validatorSubscription = validator.start()
                     .subscribe(this::setStatus)
         }
     }
 
+    override fun isRunning(): Boolean {
+        return true
+    }
+
+    override fun stop() {
+        validatorSubscription?.dispose()
+        validatorSubscription = null
+    }
+
     open fun createHead(): EthereumHead {
         return if (ethereumWs != null) {
             EthereumWsHead(ethereumWs)

src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt

@@ -12,6 +12,8 @@ import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.*
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.TopicProcessor
@@ -28,21 +30,20 @@ open class GrpcUpstream(
         private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val objectMapper: ObjectMapper,
         private val targets: CallMethods
-): DefaultUpstream() {
+): DefaultUpstream(), Lifecycle {
-
 
     private val log = LoggerFactory.getLogger(GrpcUpstream::class.java)
 
     private val options = UpstreamsConfig.Options.getDefaults()
     private val headBlock = AtomicReference<BlockJson<TransactionId>>(null)
     private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
-    private val status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)
     private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
     private val head = Head(this)
-    private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
     private val supportedMethods = HashSet<String>()
     private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
 
+    private var headSubscription: Disposable? = null
+
     open fun createApi(matcher: Selector.Matcher): EthereumApi {
         val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher))
         return EthereumApi(rpcClient, objectMapper, chain, targets).let {
@@ -51,7 +52,7 @@ open class GrpcUpstream(
         }
     }
 
-    open fun connect() {
+    override fun start() {
         val chainRef = Common.Chain.newBuilder()
                 .setTypeValue(chain.id)
                 .build()
@@ -64,11 +65,22 @@ open class GrpcUpstream(
 
         val flux = client.subscribeHead(chainRef)
                 .compose(GrpcRetry.ManyToMany.retryAfter(retry, Duration.ofSeconds(5)))
-        subscribe(flux)
+        observeHead(flux)
     }
 
-    internal fun subscribe(flux: Flux<BlockchainOuterClass.ChainHead>) {
+    override fun isRunning(): Boolean {
-        flux.map { value ->
+        return headSubscription != null
+    }
+
+
+    override fun stop() {
+        headSubscription?.dispose()
+        headSubscription = null
+    }
+
+
+    internal fun observeHead(flux: Flux<BlockchainOuterClass.ChainHead>) {
+        headSubscription = flux.map { value ->
                     val block = BlockJson<TransactionId>()
                     block.number = value.height
                     block.totalDifficulty = BigInteger(1, value.weight.toByteArray())

src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt

@@ -94,7 +94,7 @@ class GrpcUpstreams(
                 val created = GrpcUpstream(chain, client!!, objectMapper, upstreams.targetFor(chain))
                 known[chain] = created
                 upstreams.addUpstream(chain, created)
-                created.connect()
+                created.start()
                 created
             } else {
                 current

src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.toFlux
@@ -14,16 +15,25 @@ import java.time.Duration
 class HeadLagObserver (
         private val master: EthereumHead,
         private val followers: Collection<Upstream>
-): Closeable {
+): Lifecycle {
 
     private val log = LoggerFactory.getLogger(HeadLagObserver::class.java)
 
     private var current: Disposable? = null
 
-    fun start() {
+    override fun start() {
         current = subscription().subscribe { }
     }
 
+    override fun isRunning(): Boolean {
+        return current != null
+    }
+
+    override fun stop() {
+        current?.dispose()
+        current = null
+    }
+
     private fun subscription(): Flux<Unit> {
         return master.getFlux()
                 .flatMap(this::probeFollowers)
@@ -69,9 +79,4 @@ class HeadLagObserver (
         return 6
     }
 
-    override fun close() {
-        current?.dispose()
-        current = null
-    }
-
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy

@@ -56,7 +56,7 @@ class GrpcUpstreamSpec extends Specification {
         def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets)
         upstream.setLag(0)
         when:
-        upstream.connect()
+        upstream.start()
         def h = upstream.head.head.block(Duration.ofSeconds(1))
         then:
         callData.chain == Chain.ETHEREUM.id
@@ -112,7 +112,7 @@ class GrpcUpstreamSpec extends Specification {
         def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets)
         upstream.setLag(0)
         when:
-        upstream.connect()
+        upstream.start()
         finished.get()
         def h = upstream.head.head.block(Duration.ofSeconds(1))
         then:
@@ -169,7 +169,7 @@ class GrpcUpstreamSpec extends Specification {
         def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets)
         upstream.setLag(0)
         when:
-        upstream.connect()
+        upstream.start()
         finished.get()
         def h = upstream.head.head.block(Duration.ofSeconds(1))
         then: