diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index 63640e3a..6827735e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/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 { val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) - return (up as EthereumMultistream) + return (up as EthereumLikeMultistream) .getSubscribe() .subscribe(method, params) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index f9557cb7..62df3041 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -38,7 +38,7 @@ open class EthereumMultistream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)) { +) : Multistream(chain, upstreams as MutableList, 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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt index 69487b1e..cb160fa3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt index 3c59b7de..252276a7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt index 1309e4c2..ce99e941 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt +++ b/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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt index 0bcc1f71..f5d1b052 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt index f1f2e821..fc0d7285 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt index 5461072e..374193e0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index e81b28e2..90679289 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -38,7 +38,7 @@ open class EthereumPosMultistream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)) { +) : Multistream(chain, upstreams as MutableList, 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 { - throw Error("Does not supports subscription for PoS ethereum") + override fun getSubscribe(): EthereumSubscribe { + return subscribe } override fun getFeeEstimation(): ChainFees {