problem: doesn't start head subscription for a single grpc based upstream; no validation for second bitcoin upstream

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt

@@ -72,13 +72,18 @@ open class BitcoinMultistream(
         val head = if (upstreams.size == 1) {
             val upstream = upstreams.first()
             upstream.setLag(0)
-            upstream.getHead()
+            upstream.getHead().apply {
+                if (this is Lifecycle) {
+                    this.start()
+                }
+            }
         } else {
             val newHead = MergedHead(upstreams.map { it.getHead() }).apply {
                 this.start()
             }
             val lagObserver = BitcoinHeadLagObserver(newHead, upstreams)
             this.lagObserver = lagObserver
+            lagObserver.start()
             newHead
         }
         onHeadUpdated(head)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt

@@ -95,16 +95,19 @@ open class EthereumMultistream(
         val head = if (upstreams.size == 1) {
             val upstream = upstreams.first()
             upstream.setLag(0)
-            upstream.getHead()
+            upstream.getHead().apply {
+                if (this is Lifecycle) {
+                    this.start()
+                }
+            }
         } else {
             val heads = upstreams.map { it.getHead() }
             val newHead = MergedHead(heads).apply {
                 this.start()
             }
-            val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream>).apply {
+            val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream>)
-                this.start()
-            }
             this.lagObserver = lagObserver
+            lagObserver.start()
             newHead
         }
         onHeadUpdated(head)

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

@@ -43,6 +43,7 @@ class EthereumRpcHead(
     private var refreshSubscription: Disposable? = null
 
     override fun start() {
+        refreshSubscription?.dispose()
         val base = Flux.interval(interval)
             .publishOn(scheduler)
             .flatMap {