refactor ethereum subscription to support ethereum pos

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt

@@ -20,6 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.grpc.BlockchainType
 import io.emeraldpay.grpc.Chain
@@ -83,7 +84,7 @@ open class NativeSubscribe(
 
     open fun subscribe(chain: Chain, method: String, params: Any?): Flux<out Any> {
         val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
-        return (up as EthereumMultistream)
+        return (up as EthereumLikeMultistream)
             .getSubscribe()
             .subscribe(method, params)
     }

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

@@ -38,7 +38,7 @@ open class EthereumMultistream(
     chain: Chain,
     val upstreams: MutableList<EthereumUpstream>,
     caches: Caches
-) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)) {
+) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)), EthereumLikeMultistream {
 
     companion object {
         private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)
@@ -80,7 +80,7 @@ open class EthereumMultistream(
         return super.isRunning() || reader.isRunning
     }
 
-    open fun getReader(): EthereumReader {
+    override fun getReader(): EthereumReader {
         return reader
     }
 
@@ -138,7 +138,7 @@ open class EthereumMultistream(
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead()))
     }
 
-    open fun getSubscribe(): EthereumSubscribe {
+    override fun getSubscribe(): EthereumSubscribe {
         return subscribe
     }
 

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

@@ -9,7 +9,7 @@ import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 
 open class EthereumSubscribe(
-    val upstream: EthereumMultistream
+    val upstream: EthereumLikeMultistream
 ) {
 
     companion object {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt

@@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
@@ -32,7 +33,7 @@ import kotlin.concurrent.withLock
 import kotlin.concurrent.write
 
 class ConnectBlockUpdates(
-    private val upstream: EthereumMultistream
+    private val upstream: EthereumLikeMultistream
 ) {
 
     companion object {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.domain.Address
@@ -25,7 +26,7 @@ import reactor.core.publisher.Flux
 import java.util.function.Function
 
 open class ConnectLogs(
-    upstream: EthereumMultistream,
+    upstream: EthereumLikeMultistream,
     private val connectBlockUpdates: ConnectBlockUpdates,
 ) {
 
@@ -36,7 +37,7 @@ open class ConnectLogs(
         private val TOPIC_COMPARATOR = HexDataComparator()
     }
 
-    constructor(upstream: EthereumMultistream) : this(upstream, ConnectBlockUpdates(upstream))
+    constructor(upstream: EthereumLikeMultistream) : this(upstream, ConnectBlockUpdates(upstream))
 
     private val produceLogs = ProduceLogs(upstream)
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
 import org.slf4j.LoggerFactory
@@ -28,7 +29,7 @@ import kotlin.concurrent.withLock
  * Connects/reconnects to the upstream to produce NewHeads messages
  */
 class ConnectNewHeads(
-    private val upstream: EthereumMultistream
+    private val upstream: EthereumLikeMultistream
 ) {
 
     companion object {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt

@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
@@ -24,7 +25,7 @@ import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 
 class ConnectSyncing(
-    private val upstream: EthereumMultistream
+    private val upstream: EthereumLikeMultistream
 ) {
 
     companion object {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.hex.HexData
@@ -38,7 +39,7 @@ class ProduceLogs(
         private val log = LoggerFactory.getLogger(ProduceLogs::class.java)
     }
 
-    constructor(upstream: EthereumMultistream) : this(upstream.getReader().receipts())
+    constructor(upstream: EthereumLikeMultistream) : this(upstream.getReader().receipts())
 
     private val objectMapper = Global.objectMapper
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -38,7 +38,7 @@ open class EthereumPosMultistream(
     chain: Chain,
     val upstreams: MutableList<EthereumPosUpstream>,
     caches: Caches
-) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)) {
+) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)), EthereumLikeMultistream {
 
     companion object {
         private val log = LoggerFactory.getLogger(EthereumPosMultistream::class.java)
@@ -48,6 +48,8 @@ open class EthereumPosMultistream(
 
     private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
     private val feeEstimation = EthereumPriorityFees(this, reader, 256)
+    private val subscribe = EthereumSubscribe(this)
+
     init {
         this.init()
     }
@@ -73,7 +75,7 @@ open class EthereumPosMultistream(
         return super.isRunning() || reader.isRunning
     }
 
-    open fun getReader(): EthereumReader {
+    override fun getReader(): EthereumReader {
         return reader
     }
 
@@ -131,8 +133,8 @@ open class EthereumPosMultistream(
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead()))
     }
 
-    open fun getSubscribe(): EthereumSubscribe {
+    override fun getSubscribe(): EthereumSubscribe {
-        throw Error("Does not supports subscription for PoS ethereum")
+        return subscribe
     }
 
     override fun getFeeEstimation(): ChainFees {