solution: refactor upstreams to make them blockchain specific

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

@@ -17,38 +17,34 @@ package io.emeraldpay.dshackle.upstream
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
-import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumHeadMerge
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Mono
-import java.lang.IllegalStateException
-import java.time.Duration
 
-open class ChainUpstreams (
+/**
+ * General interface to upstream(s) to a single chain
+ */
+abstract class ChainUpstreams<U : UpstreamApi, B>(
         val chain: Chain,
-        private val upstreams: MutableList<Upstream>,
+        private val upstreams: MutableList<Upstream<U, B>>,
         caches: Caches,
         objectMapper: ObjectMapper
-) : AggregatedUpstream(objectMapper, caches), Lifecycle {
+) : AggregatedUpstream<U, B>(objectMapper, caches), Lifecycle {
 
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0
-    private var head: EthereumHead? = null
-    private var lagObserver: HeadLagObserver? = null
+    protected var lagObserver: HeadLagObserver<U, B>? = null
     private var subscription: Disposable? = null
 
-    init {
-        if (upstreams.size > 0) {
-            head = updateHead()
-            onUpstreamsUpdated()
-        }
+    open fun init() {
+        onUpstreamsUpdated()
     }
 
+    abstract fun updateHead(): Head<B>
+    abstract fun setHead(head: Head<B>)
+
     override fun getId(): String {
         return "!all:${chain.chainCode}"
     }
@@ -68,7 +64,7 @@ open class ChainUpstreams (
         super.stop()
         subscription?.dispose()
         subscription = null
-        head?.let {
+        getHead().let {
             if (it is Lifecycle) {
                 it.stop()
             }
@@ -76,50 +72,24 @@ open class ChainUpstreams (
         lagObserver?.stop()
     }
 
-    internal fun updateHead(): EthereumHead {
-        head?.let {
-            if (it is Lifecycle) {
-                it.stop()
-            }
-        }
-        lagObserver?.stop()
-        lagObserver = null
-        val head = if (upstreams.size == 1) {
-            val upstream = upstreams.first()
-            upstream.setLag(0)
-            upstream.getHead()
-        } else {
-            val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }).apply {
-                this.start()
-            }
-            val lagObserver = HeadLagObserver(newHead, upstreams).apply {
-                this.start()
-            }
-            this.lagObserver = lagObserver
-            newHead
-        }
-        onHeadUpdated(head)
-        return head
-    }
-
-    override fun getAll(): List<Upstream> {
+    override fun getAll(): List<Upstream<U, B>> {
         return upstreams
     }
 
-    override fun addUpstream(upstream: Upstream) {
+    override fun addUpstream(upstream: Upstream<U, B>) {
         upstreams.add(upstream)
-        head = updateHead()
+        setHead(updateHead())
         onUpstreamsUpdated()
     }
 
     fun removeUpstream(id: String) {
         if (upstreams.removeIf { it.getId() == id }) {
-            head = updateHead()
+            setHead(updateHead())
             onUpstreamsUpdated()
         }
     }
 
-    override fun getApis(matcher: Selector.Matcher): ApiSource {
+    override fun getApis(matcher: Selector.Matcher): ApiSource<U> {
         val i = seq++
         if (seq >= Int.MAX_VALUE / 2) {
             seq = 0
@@ -127,15 +97,11 @@ open class ChainUpstreams (
         return FilteredApis(upstreams, matcher, i)
     }
 
-    override fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi> {
+    override fun getApi(matcher: Selector.Matcher): Mono<U> {
         val apis = getApis(matcher)
         apis.request(1)
         return Mono.from(apis)
-                .switchIfEmpty(Mono.error<DirectEthereumApi>(Exception("No API available")))
-    }
-
-    override fun getHead(): EthereumHead {
-        return head!!
+                .switchIfEmpty(Mono.error<U>(Exception("No API available")))
     }
 
     override fun setLag(lag: Long) {
@@ -145,28 +111,5 @@ open class ChainUpstreams (
         return 0
     }
 
-    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
-        return upstreams.flatMap { it.getLabels() }
-    }
-
-    fun printStatus() {
-        var height: Long? = null
-        try {
-            height = head!!.getFlux().next().block(Duration.ofSeconds(1))?.number
-        } catch (e: IllegalStateException) {
-            //timout
-        } catch (e: Exception) {
-            log.warn("Head processing error: ${e.javaClass} ${e.message}")
-        }
-        val statuses = upstreams.map { it.getStatus() }
-                .groupBy { it }
-                .map { "${it.key.name}/${it.value.size}" }
-                .joinToString(",")
-        val lag = upstreams.map { it.getLag() }
-                .joinToString(", ")
-
-        log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]")
-    }
-
-
+    abstract fun printStatus()
 }