From bbc21684ae7fd9cf149bbf68dc758b4012c64ff7 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 14 May 2020 00:12:38 -0400 Subject: [PATCH] solution: refactoring, more clean class names --- build.gradle | 1 - .../io/emeraldpay/dshackle/rpc/Describe.kt | 6 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 10 +- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 8 +- .../dshackle/rpc/SubscribeStatus.kt | 8 +- .../dshackle/rpc/TrackBitcoinAddress.kt | 19 ++- .../emeraldpay/dshackle/rpc/TrackBitcoinTx.kt | 25 ++-- .../dshackle/rpc/TrackEthereumAddress.kt | 16 +-- .../dshackle/rpc/TrackEthereumTx.kt | 16 +-- .../dshackle/startup/ConfiguredUpstreams.kt | 8 +- .../dshackle/upstream/ChainUpstreams.kt | 136 ------------------ ...streams.kt => CurrentMultistreamHolder.kt} | 24 ++-- .../{AggregatedUpstream.kt => Multistream.kt} | 98 ++++++++++++- .../{Upstreams.kt => MultistreamHolder.kt} | 11 +- ...hainUpstreams.kt => BitcoinMultistream.kt} | 8 +- .../upstream/bitcoin/BitcoinReader.kt | 2 +- .../upstream/bitcoin/CachingMempoolData.kt | 2 +- ...hainUpstream.kt => EthereumMultistream.kt} | 6 +- .../upstream/ethereum/EthereumReader.kt | 7 +- .../upstream/grpc/EthereumGrpcUpstream.kt | 3 - .../dshackle/upstream/grpc/GrpcUpstreams.kt | 5 +- .../dshackle/rpc/NativeCallSpec.groovy | 31 ++-- .../dshackle/rpc/StreamHeadSpec.groovy | 6 +- .../rpc/TrackBitcoinAddressSpec.groovy | 38 +++-- .../dshackle/rpc/TrackBitcoinTxSpec.groovy | 39 +++-- .../rpc/TrackEthereumAddressSpec.groovy | 17 +-- .../dshackle/rpc/TrackEthereumTxSpec.groovy | 22 +-- ...ck.groovy => MultistreamHolderMock.groovy} | 32 ++--- .../dshackle/test/TestingCommons.groovy | 10 +- ...vy => CurrentMultistreamHolderSpec.groovy} | 11 +- ...reamSpec.groovy => MultistreamSpec.groovy} | 6 +- .../ethereum/EthereumReaderSpec.groovy | 17 +-- .../grpc/EthereumGrpcUpstreamSpec.groovy | 2 - 33 files changed, 275 insertions(+), 375 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/{CurrentUpstreams.kt => CurrentMultistreamHolder.kt} (85%) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/{AggregatedUpstream.kt => Multistream.kt} (62%) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/{Upstreams.kt => MultistreamHolder.kt} (77%) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/{BitcoinChainUpstreams.kt => BitcoinMultistream.kt} (93%) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumChainUpstream.kt => EthereumMultistream.kt} (94%) rename src/test/groovy/io/emeraldpay/dshackle/test/{UpstreamsMock.groovy => MultistreamHolderMock.groovy} (72%) rename src/test/groovy/io/emeraldpay/dshackle/upstream/{CurrentUpstreamsSpec.groovy => CurrentMultistreamHolderSpec.groovy} (87%) rename src/test/groovy/io/emeraldpay/dshackle/upstream/{AggregatedUpstreamSpec.groovy => MultistreamSpec.groovy} (86%) diff --git a/build.gradle b/build.gradle index 37f46840..2ec4366d 100644 --- a/build.gradle +++ b/build.gradle @@ -81,7 +81,6 @@ dependencies { implementation "io.infinitape:etherjar-hex:$etherjarVersion" implementation "io.infinitape:etherjar-rpc-http:$etherjarVersion" implementation "io.infinitape:etherjar-rpc-ws:$etherjarVersion" - implementation "io.infinitape:etherjar-rpc-emerald:$etherjarVersion" implementation "io.infinitape:etherjar-tx:$etherjarVersion" implementation 'org.bitcoinj:bitcoinj-core:0.15.8' diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 82ebbcb0..45291ccf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -29,15 +29,15 @@ import reactor.core.publisher.Mono @Service class Describe( - @Autowired private val upstreams: Upstreams, + @Autowired private val multistreamHolder: MultistreamHolder, @Autowired private val subscribeStatus: SubscribeStatus ) { fun describe(requestMono: Mono): Mono { return requestMono.map { _ -> val resp = BlockchainOuterClass.DescribeResponse.newBuilder() - upstreams.getAvailable().forEach { chain -> - upstreams.getUpstream(chain)?.let { chainUpstreams -> + multistreamHolder.getAvailable().forEach { chain -> + multistreamHolder.getUpstream(chain)?.let { chainUpstreams -> val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll()) val targets = chainUpstreams.getMethods().getSupportedMethods() val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 6e2ec887..b8ca8bbe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -37,7 +37,7 @@ import java.lang.Exception @Service open class NativeCall( - @Autowired private val upstreams: Upstreams, + @Autowired private val multistreamHolder: MultistreamHolder, @Autowired private val objectMapper: ObjectMapper ) { @@ -88,17 +88,17 @@ open class NativeCall( return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) } - if (!upstreams.isAvailable(chain)) { + if (!multistreamHolder.isAvailable(chain)) { return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) } - val upstream = upstreams.getUpstream(chain) + val upstream = multistreamHolder.getUpstream(chain) ?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain))) return prepareCall(request, upstream) } - fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux> { + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: Multistream): Flux> { return request.itemsList.toFlux().map { val method = it.method val params = it.payload.toStringUtf8() @@ -202,7 +202,7 @@ open class NativeCall( } open class CallContext(val id: Int, - val upstream: AggregatedUpstream, + val upstream: Multistream, val matcher: Selector.Matcher, val callQuorum: CallQuorum, val payload: T) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 91d9bece..620efcb2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -21,10 +21,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -33,7 +31,7 @@ import reactor.core.publisher.Mono @Service class StreamHead( - @Autowired private val upstreams: Upstreams + @Autowired private val multistreamHolder: MultistreamHolder ) { private val log = LoggerFactory.getLogger(StreamHead::class.java) @@ -42,7 +40,7 @@ class StreamHead( return requestMono.map { request -> Chain.byId(request.type.number) }.flatMapMany { chain -> - val up = upstreams.getUpstream(chain) + val up = multistreamHolder.getUpstream(chain) ?: return@flatMapMany Flux.error(Exception("Unavailable chain: $chain")) up.getHead() .getFlux() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index 9ad74488..f22b2b17 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -27,13 +27,13 @@ import reactor.core.publisher.Mono @Service class SubscribeStatus( - @Autowired private val upstreams: Upstreams + @Autowired private val multistreamHolder: MultistreamHolder ) { fun subscribeStatus(requestMono: Mono): Flux { return requestMono.flatMapMany { - val ups = upstreams.getAvailable().mapNotNull { chain -> - val chainUpstream = upstreams.getUpstream(chain) + val ups = multistreamHolder.getAvailable().mapNotNull { chain -> + val chainUpstream = multistreamHolder.getUpstream(chain) chainUpstream?.observeStatus()?.map { avail -> ChainSubscription(chain, chainUpstream, avail) } @@ -60,6 +60,6 @@ class SubscribeStatus( .build() } - class ChainSubscription(val chain: Chain, val up: AggregatedUpstream, val avail: UpstreamAvailability) + class ChainSubscription(val chain: Chain, val up: Multistream, val avail: UpstreamAvailability) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt index 1296e926..dededb4b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt @@ -19,9 +19,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.SilentException -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -35,7 +34,7 @@ import kotlin.collections.HashMap @Service class TrackBitcoinAddress( - @Autowired private val upstreams: Upstreams + @Autowired private val multistreamHolder: MultistreamHolder ) : TrackAddress { companion object { @@ -43,7 +42,7 @@ class TrackBitcoinAddress( } override fun isSupported(chain: Chain): Boolean { - return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && upstreams.isAvailable(chain) + return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain) } fun allAddresses(request: BlockchainOuterClass.BalanceRequest): List? { @@ -63,7 +62,7 @@ class TrackBitcoinAddress( } } - fun requestBalances(chain: Chain, api: BitcoinChainUpstreams, addresses: List): Flux { + fun requestBalances(chain: Chain, api: BitcoinMultistream, addresses: List): Flux { return api.getReader().listUnspent() .flatMapMany { unspents -> val result = getTotal(chain, addresses, unspents) @@ -73,7 +72,7 @@ class TrackBitcoinAddress( override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) - val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java) + val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java) ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address")) if (addresses.isEmpty()) { @@ -119,9 +118,9 @@ class TrackBitcoinAddress( override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) - println("up: ${upstreams.getUpstream(chain)}") - println("up cast: ${upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)}") - val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java) + println("up: ${multistreamHolder.getUpstream(chain)}") + println("up cast: ${multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)}") + val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java) ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address")) if (addresses.isEmpty()) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt index 72254d9b..8199ff41 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt @@ -20,9 +20,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.SilentException -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory @@ -38,7 +37,7 @@ import kotlin.math.min @Service class TrackBitcoinTx( - @Autowired private val upstreams: Upstreams + @Autowired private val multistreamHolder: MultistreamHolder ) : TrackTx { companion object { @@ -46,12 +45,12 @@ class TrackBitcoinTx( } override fun isSupported(chain: Chain): Boolean { - return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && upstreams.isAvailable(chain) + return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain) } override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux { val chain = Chain.byId(request.chainValue) - val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java) + val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) val txid = request.txId val confirmations = max(min(1, request.confirmationLimit), 12) @@ -61,7 +60,7 @@ class TrackBitcoinTx( }.map(this::asProto) } - fun subscribe(chain: Chain, upstream: BitcoinChainUpstreams, txid: String): Flux { + fun subscribe(chain: Chain, upstream: BitcoinMultistream, txid: String): Flux { return loadExisting(upstream, txid) .flatMapMany { status -> if (status.mined) { @@ -83,7 +82,7 @@ class TrackBitcoinTx( } } - fun continueWithMined(upstream: BitcoinChainUpstreams, status: TxStatus): Flux { + fun continueWithMined(upstream: BitcoinMultistream, status: TxStatus): Flux { return upstream.getReader().getBlock(status.blockHash!!) .map { block -> TxStatus(status.txid, true, ExtractBlock.getHeight(block), true, status.blockHash, ExtractBlock.getTime(block), ExtractBlock.getDifficulty(block)) @@ -92,7 +91,7 @@ class TrackBitcoinTx( } } - fun untilFound(chain: Chain, upstream: BitcoinChainUpstreams, txid: String): Flux { + fun untilFound(chain: Chain, upstream: BitcoinMultistream, txid: String): Flux { return Flux.interval(Duration.ofSeconds(1)) .take(Duration.ofMinutes(10)) .flatMap { loadMempool(upstream, txid) } @@ -103,20 +102,20 @@ class TrackBitcoinTx( } } - fun untilMined(upstream: BitcoinChainUpstreams, tx: TxStatus): Mono { + fun untilMined(upstream: BitcoinMultistream, tx: TxStatus): Mono { return upstream.getHead().getFlux().flatMap { loadExisting(upstream, tx.txid) .filter { it.mined } }.single() } - fun withConfirmations(upstream: BitcoinChainUpstreams, tx: TxStatus): Flux { + fun withConfirmations(upstream: BitcoinMultistream, tx: TxStatus): Flux { return upstream.getHead().getFlux().map { tx.withHead(it.height) } } - fun loadExisting(api: BitcoinChainUpstreams, txid: String): Mono { + fun loadExisting(api: BitcoinMultistream, txid: String): Mono { val mined = api.getReader().getTx(txid) return mined.map { val block = it["blockhash"] as String? @@ -124,7 +123,7 @@ class TrackBitcoinTx( } } - fun loadMempool(upstream: BitcoinChainUpstreams, txid: String): Mono { + fun loadMempool(upstream: BitcoinMultistream, txid: String): Mono { val mempool = upstream.getReader().getMempool().get() return mempool.map { if (it.contains(txid)) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt index 4d364f14..b953d448 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt @@ -21,8 +21,8 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.SilentException -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.Wei @@ -34,13 +34,13 @@ import reactor.core.publisher.Mono @Service class TrackEthereumAddress( - @Autowired private val upstreams: Upstreams + @Autowired private val multistreamHolder: MultistreamHolder ) : TrackAddress { private val log = LoggerFactory.getLogger(TrackEthereumAddress::class.java) override fun isSupported(chain: Chain): Boolean { - return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain) + return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && multistreamHolder.isAvailable(chain) } override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux { @@ -51,7 +51,7 @@ class TrackEthereumAddress( override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) - val head = upstreams.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty() + val head = multistreamHolder.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty() val balances = initAddress(request) .flatMap { tracked -> val current = getBalance(tracked) @@ -86,14 +86,14 @@ class TrackEthereumAddress( } } - fun getUpstream(chain: Chain): EthereumChainUpstream { - return upstreams.getUpstream(chain)?.cast(EthereumChainUpstream::class.java) + fun getUpstream(chain: Chain): EthereumMultistream { + return multistreamHolder.getUpstream(chain)?.cast(EthereumMultistream::class.java) ?: throw SilentException.UnsupportedBlockchain(chain) } private fun initAddress(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) - if (!upstreams.isAvailable(chain)) { + if (!multistreamHolder.isAvailable(chain)) { return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) } if (request.asset.code?.toLowerCase() != "ether") { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt index 8cd6d010..9ef50393 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt @@ -23,8 +23,8 @@ import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.TxId -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -48,7 +48,7 @@ import kotlin.math.min @Service class TrackEthereumTx( - @Autowired private val upstreams: Upstreams + @Autowired private val multistreamHolder: MultistreamHolder ) : TrackTx { companion object { @@ -63,7 +63,7 @@ class TrackEthereumTx( private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java) override fun isSupported(chain: Chain): Boolean { - return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain) + return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && multistreamHolder.isAvailable(chain) } override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux { @@ -83,12 +83,12 @@ class TrackEthereumTx( } - fun getUpstream(chain: Chain): EthereumChainUpstream { - return upstreams.getUpstream(chain)?.cast(EthereumChainUpstream::class.java) + fun getUpstream(chain: Chain): EthereumMultistream { + return multistreamHolder.getUpstream(chain)?.cast(EthereumMultistream::class.java) ?: throw SilentException.UnsupportedBlockchain(chain) } - fun subscribe(base: TxDetails, up: EthereumChainUpstream): Flux { + fun subscribe(base: TxDetails, up: EthereumMultistream): Flux { var latestTx = base val untilFound = Mono.just(latestTx) @@ -213,7 +213,7 @@ class TrackEthereumTx( } } - fun updateFromBlock(upstream: EthereumChainUpstream, tx: TxDetails, blockTx: TransactionJson): Mono { + fun updateFromBlock(upstream: EthereumMultistream, tx: TxDetails, blockTx: TransactionJson): Mono { return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) { val updated = tx.withStatus( blockHash = blockTx.blockHash, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 907181c1..8a62f67b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.CurrentUpstreams +import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods @@ -45,7 +45,7 @@ import kotlin.collections.HashMap @Repository open class ConfiguredUpstreams( @Autowired private val objectMapper: ObjectMapper, - @Autowired private val currentUpstreams: CurrentUpstreams, + @Autowired private val currentUpstreams: CurrentMultistreamHolder, @Autowired private val fileResolver: FileResolver, @Autowired private val config: UpstreamsConfig, @Autowired private val cachesFactory: CachesFactory @@ -201,8 +201,7 @@ open class ConfiguredUpstreams( endpoint.port ?: 2449, objectMapper, endpoint.auth, - fileResolver, - cachesFactory + fileResolver ).apply { timeout = options.timeout } @@ -214,7 +213,6 @@ open class ConfiguredUpstreams( .subscribe(currentUpstreams::update) } - private fun buildHttpClient(config: UpstreamsConfig.Upstream): JsonRpcHttpClient? { val conn = config.connection!! val urls = ArrayList() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt deleted file mode 100644 index b8a65707..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ /dev/null @@ -1,136 +0,0 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * Copyright (c) 2019 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -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 - -/** - * General interface to upstream(s) to a single chain - */ -abstract class ChainUpstreams( - val chain: Chain, - private val upstreams: MutableList, - caches: Caches -) : AggregatedUpstream(caches), Lifecycle { - - private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) - private var seq = 0 - protected var lagObserver: HeadLagObserver? = null - private var subscription: Disposable? = null - - open fun init() { - onUpstreamsUpdated() - } - - abstract fun updateHead(): Head - abstract fun setHead(head: Head) - - override fun getId(): String { - return "!all:${chain.chainCode}" - } - - override fun isRunning(): Boolean { - return subscription != null - } - - override fun start() { - super.start() - subscription = observeStatus() - .distinctUntilChanged() - .subscribe { printStatus() } - } - - override fun stop() { - super.stop() - subscription?.dispose() - subscription = null - getHead().let { - if (it is Lifecycle) { - it.stop() - } - } - lagObserver?.stop() - } - - override fun getAll(): List { - return upstreams - } - - override fun addUpstream(upstream: Upstream) { - upstreams.add(upstream) - setHead(updateHead()) - onUpstreamsUpdated() - } - - fun removeUpstream(id: String) { - if (upstreams.removeIf { it.getId() == id }) { - setHead(updateHead()) - onUpstreamsUpdated() - } - } - - override fun getApiSource(matcher: Selector.Matcher): ApiSource { - val i = seq++ - if (seq >= Int.MAX_VALUE / 2) { - seq = 0 - } - return FilteredApis(upstreams, matcher, i) - } - - override fun getDirectApi(matcher: Selector.Matcher): Mono> { - val apis = getApiSource(matcher) - apis.request(1) - return Mono.from(apis) - .switchIfEmpty(Mono.error(Exception("No API available"))) - } - - override fun setLag(lag: Long) { - } - - override fun getLag(): Long { - return 0 - } - - fun printStatus() { - var height: Long? = null - try { - height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.height - } 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]") - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt similarity index 85% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index 484fa68a..efe076e6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -21,12 +21,12 @@ import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.startup.UpstreamChange -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory @@ -42,14 +42,14 @@ import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @Repository -class CurrentUpstreams( +class CurrentMultistreamHolder( @Autowired private val objectMapper: ObjectMapper, @Autowired private val cachesFactory: CachesFactory -): Upstreams { +) : MultistreamHolder { - private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java) + private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java) - private val chainMapping = ConcurrentHashMap() + private val chainMapping = ConcurrentHashMap() private val chainsBus = TopicProcessor.create() private val callTargets = HashMap() private val updateLock = ReentrantLock() @@ -60,17 +60,17 @@ class CurrentUpstreams( when (BlockchainType.fromBlockchain(chain)) { BlockchainType.ETHEREUM -> { val up = change.upstream.cast(EthereumUpstream::class.java) - val current = chainMapping[chain] as ChainUpstreams? + val current = chainMapping[chain] as Multistream? val factory = Callable { - EthereumChainUpstream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams + EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as Multistream } processUpdate(change, up, current, factory) } BlockchainType.BITCOIN -> { val up = change.upstream.cast(BitcoinUpstream::class.java) - val current = chainMapping[chain] as ChainUpstreams? + val current = chainMapping[chain] as Multistream? val factory = Callable { - BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams + BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as Multistream } processUpdate(change, up, current, factory) } @@ -81,7 +81,7 @@ class CurrentUpstreams( } } - fun processUpdate(change: UpstreamChange, up: Upstream, current: ChainUpstreams?, factory: Callable) { + fun processUpdate(change: UpstreamChange, up: Upstream, current: Multistream?, factory: Callable) { val chain = change.chain if (change.type == UpstreamChange.ChangeType.REMOVED) { current?.removeUpstream(up.getId()) @@ -109,7 +109,7 @@ class CurrentUpstreams( } } - override fun getUpstream(chain: Chain): AggregatedUpstream? { + override fun getUpstream(chain: Chain): Multistream? { return chainMapping[chain] } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt similarity index 62% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 51b30fd4..5c9feb39 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.upstream -import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.* import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader @@ -24,6 +23,8 @@ import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -38,33 +39,70 @@ import kotlin.concurrent.withLock /** * Aggregation of multiple upstreams responding to a single blockchain */ -abstract class AggregatedUpstream( +abstract class Multistream( + val chain: Chain, + private val upstreams: MutableList, val caches: Caches ) : Upstream, Lifecycle { + companion object { + private val log = LoggerFactory.getLogger(Multistream::class.java) + } + private var cacheSubscription: Disposable? = null private val reconfigLock = ReentrantLock() private var callMethods: CallMethods? = null + private var seq = 0 + protected var lagObserver: HeadLagObserver? = null + private var subscription: Disposable? = null + + open fun init() { + onUpstreamsUpdated() + } /** * Get list of all underlying upstreams */ - abstract fun getAll(): List + fun getAll(): List { + return upstreams + } /** * Add an upstream */ - abstract fun addUpstream(upstream: Upstream) + fun addUpstream(upstream: Upstream) { + upstreams.add(upstream) + setHead(updateHead()) + onUpstreamsUpdated() + } + + fun removeUpstream(id: String) { + if (upstreams.removeIf { it.getId() == id }) { + setHead(updateHead()) + onUpstreamsUpdated() + } + } /** * Get a source for direct APIs */ - abstract fun getApiSource(matcher: Selector.Matcher): ApiSource + fun getApiSource(matcher: Selector.Matcher): ApiSource { + val i = seq++ + if (seq >= Int.MAX_VALUE / 2) { + seq = 0 + } + return FilteredApis(upstreams, matcher, i) + } /** * Finds an API that executed directly on a remote. */ - abstract fun getDirectApi(matcher: Selector.Matcher): Mono> + fun getDirectApi(matcher: Selector.Matcher): Mono> { + val apis = getApiSource(matcher) + apis.request(1) + return Mono.from(apis) + .switchIfEmpty(Mono.error(Exception("No API available"))) + } /** * Finds an API that leverages caches and other optimizations/transformations of the request. @@ -109,11 +147,22 @@ abstract class AggregatedUpstream( } override fun start() { + subscription = observeStatus() + .distinctUntilChanged() + .subscribe { printStatus() } } override fun stop() { cacheSubscription?.dispose() cacheSubscription = null + subscription?.dispose() + subscription = null + getHead().let { + if (it is Lifecycle) { + it.stop() + } + } + lagObserver?.stop() } fun onHeadUpdated(head: Head) { @@ -125,6 +174,43 @@ abstract class AggregatedUpstream( } } + abstract fun updateHead(): Head + abstract fun setHead(head: Head) + + override fun getId(): String { + return "!all:${chain.chainCode}" + } + + override fun isRunning(): Boolean { + return subscription != null + } + + override fun setLag(lag: Long) { + } + + override fun getLag(): Long { + return 0 + } + + fun printStatus() { + var height: Long? = null + try { + height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.height + } catch (e: java.lang.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]") + } + // -------------------------------------------------------------------------------------------------------- class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt similarity index 77% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt index 91db9cda..e9f85f68 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt @@ -16,16 +16,15 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -interface Upstreams { - fun getUpstream(chain: Chain): AggregatedUpstream? +/** + * Holds Multistreams configured for a chain. + */ +interface MultistreamHolder { + fun getUpstream(chain: Chain): Multistream? fun getAvailable(): List fun observeChains(): Flux fun getDefaultMethods(chain: Chain): CallMethods diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt similarity index 93% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index 9b5467f0..131c3433 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -27,20 +27,18 @@ import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.publisher.Mono -open class BitcoinChainUpstreams( +open class BitcoinMultistream( chain: Chain, val upstreams: MutableList, caches: Caches, private val objectMapper: ObjectMapper -) : ChainUpstreams(chain, upstreams as MutableList, caches), Lifecycle { +) : Multistream(chain, upstreams as MutableList, caches), Lifecycle { companion object { - private val log = LoggerFactory.getLogger(BitcoinChainUpstreams::class.java) + private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) } private var head: Head? = null - - //TODO head private var reader = BitcoinReader(this, EmptyHead(), objectMapper) override fun init() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt index 78c2aa9d..86f5ad3d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt @@ -26,7 +26,7 @@ import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.cast open class BitcoinReader( - private val upstreams: BitcoinChainUpstreams, + private val upstreams: BitcoinMultistream, head: Head, private val objectMapper: ObjectMapper ) : Lifecycle { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt index 6cecfca9..d18356ba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt @@ -30,7 +30,7 @@ import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock open class CachingMempoolData( - private val upstreams: BitcoinChainUpstreams, + private val upstreams: BitcoinMultistream, private val head: Head, private val objectMapper: ObjectMapper ) : Lifecycle { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt similarity index 94% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 1d793fa8..de6d9f7b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -28,15 +28,15 @@ import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.publisher.Mono -open class EthereumChainUpstream( +open class EthereumMultistream( chain: Chain, val upstreams: MutableList, caches: Caches, private val objectMapper: ObjectMapper -) : ChainUpstreams(chain, upstreams as MutableList, caches) { +) : Multistream(chain, upstreams as MutableList, caches) { companion object { - private val log = LoggerFactory.getLogger(EthereumChainUpstream::class.java) + private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) } private var head: Head? = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt index fd381ed0..c91b14fa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt @@ -21,10 +21,8 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache import io.emeraldpay.dshackle.data.* import io.emeraldpay.dshackle.reader.* -import io.emeraldpay.dshackle.upstream.AggregatedUpstream +import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.infinitape.etherjar.domain.Address @@ -35,7 +33,6 @@ import io.infinitape.etherjar.hex.HexQuantity import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.BlockTag import io.infinitape.etherjar.rpc.json.TransactionJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory @@ -48,7 +45,7 @@ import java.util.concurrent.TimeoutException import java.util.function.Function open class EthereumReader( - private val up: AggregatedUpstream, + private val up: Multistream, private val caches: Caches, private val objectMapper: ObjectMapper ) : Lifecycle { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index b7a49598..18bd641d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -22,8 +22,6 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.Defaults -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId @@ -39,7 +37,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.* -import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 2ace9c16..4e0e3934 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -22,7 +22,6 @@ import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.FileResolver -import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.startup.UpstreamChange @@ -30,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.grpc.Chain import io.grpc.ManagedChannelBuilder import io.grpc.netty.NettyChannelBuilder -import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import io.netty.handler.ssl.* import org.apache.commons.lang3.StringUtils import org.apache.commons.lang3.exception.ExceptionUtils @@ -50,8 +48,7 @@ class GrpcUpstreams( private val port: Int, private val objectMapper: ObjectMapper, private val auth: AuthConfig.ClientTlsAuth? = null, - private val fileResolver: FileResolver, - private val cachesFactory: CachesFactory + private val fileResolver: FileResolver ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 5a01c548..6da8e1df 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -23,8 +23,7 @@ import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.NonEmptyQuorum import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcException @@ -44,7 +43,7 @@ class NativeCallSpec extends Specification { def "Quorum is applied"() { setup: def quorum = Spy(new AlwaysQuorum()) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def apiMock = TestingCommons.api() apiMock.answer("eth_test", [], "foo") @@ -67,7 +66,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def apiMock = TestingCommons.api() apiMock.answerOnce("eth_test", [], null) @@ -93,7 +92,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def apiMock = TestingCommons.api() apiMock.answerOnce("eth_test", [], null) @@ -119,7 +118,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api() @@ -143,7 +142,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) ReactorRpcClient rpcClient = Stub(ReactorRpcClient) def apiMock = TestingCommons.api() @@ -167,7 +166,7 @@ class NativeCallSpec extends Specification { def "Packs call exception into response with id"() { setup: - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) when: def resp = nativeCall.processException(new NativeCall.CallFailure(5, new IllegalArgumentException("test test"))) @@ -184,7 +183,7 @@ class NativeCallSpec extends Specification { def "Packs unknown exception into response"() { setup: - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) when: def resp = nativeCall.processException(new IllegalArgumentException("test test")) @@ -200,7 +199,7 @@ class NativeCallSpec extends Specification { def "Builds normal response"() { setup: - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def json = [jsonrpc:"2.0", id:1, result: "foo"] @@ -216,7 +215,7 @@ class NativeCallSpec extends Specification { def "Returns error for invalid chain"() { setup: - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() @@ -239,7 +238,7 @@ class NativeCallSpec extends Specification { def "Returns error for unsupported chain"() { setup: - def upstreams = Mock(Upstreams) + def upstreams = Mock(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() @@ -265,7 +264,7 @@ class NativeCallSpec extends Specification { //TODO def "Calls cache before remote"() { setup: - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def api = TestingCommons.api() def upstream = TestingCommons.aggregatedUpstream(api) @@ -284,7 +283,7 @@ class NativeCallSpec extends Specification { //TODO def "Uses cached value"() { setup: - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def upstream = TestingCommons.aggregatedUpstream(TestingCommons.api()) @@ -303,7 +302,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new AlwaysQuorum()) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def apiMock = TestingCommons.api() apiMock.answer("eth_test", [], null, 1, new TimeoutException("test 1")) @@ -329,7 +328,7 @@ class NativeCallSpec extends Specification { setup: def quorum = Spy(new BroadcastQuorum(TestingCommons.objectMapper(), 3)) - def upstreams = Stub(Upstreams) + def upstreams = Stub(MultistreamHolder) def apiMock = TestingCommons.api() apiMock.answer("eth_sendRawTransaction", ["0x1234"], diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy index e9c1e628..199629ce 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy @@ -23,7 +23,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.test.UpstreamsMock +import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash @@ -42,7 +42,7 @@ class StreamHeadSpec extends Specification { def "Errors on unavailable chain"() { setup: - def upstreams = new UpstreamsMock(Chain.ETHEREUM, Stub(EthereumUpstream)) + def upstreams = new MultistreamHolderMock(Chain.ETHEREUM, Stub(EthereumUpstream)) def streamHead = new StreamHead(upstreams) when: def flux = streamHead.add( @@ -78,7 +78,7 @@ class StreamHeadSpec extends Specification { } def upstream = new EthereumUpstreamMock(Chain.ETHEREUM, TestingCommons.api()) - def upstreams = new UpstreamsMock(Chain.ETHEREUM, upstream) + def upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstream) def streamHead = new StreamHead(upstreams) when: def flux = streamHead.add( diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy index df69d4d8..9f4a0488 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy @@ -19,18 +19,12 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.test.ReaderMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.test.UpstreamsMock -import io.emeraldpay.dshackle.upstream.AggregatedUpstream +import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinReader -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -49,7 +43,7 @@ class TrackBitcoinAddressSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-one-addr.json") def unspents = TestingCommons.objectMapper().readValue(json, List) - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) when: def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents) @@ -64,7 +58,7 @@ class TrackBitcoinAddressSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json") def unspents = TestingCommons.objectMapper().readValue(json, List) - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) when: def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents) @@ -79,7 +73,7 @@ class TrackBitcoinAddressSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json") def unspents = TestingCommons.objectMapper().readValue(json, List) - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) when: def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK", "35hK24tcLEWcgNA4JxpvbkNkoAcDGqQPsP"], unspents).sort { it.address.address } @@ -99,7 +93,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Zero for empty unspents"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) when: def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], []) @@ -114,7 +108,7 @@ class TrackBitcoinAddressSpec extends Specification { setup: def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json") def unspents = TestingCommons.objectMapper().readValue(json, List) - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) when: def total = track.getTotal(Chain.BITCOIN, ["16rCmCmbuWDhPjWTrpQGaU3EPdZF7MTdUk", "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents).sort { it.address.address } @@ -132,7 +126,7 @@ class TrackBitcoinAddressSpec extends Specification { def "One address for single provided"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def req = BlockchainOuterClass.BalanceRequest.newBuilder() .setAddress( Common.AnyAddress.newBuilder() @@ -150,7 +144,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Sorted addresses for multiple provided"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def req = BlockchainOuterClass.BalanceRequest.newBuilder() .setAddress( Common.AnyAddress.newBuilder() @@ -171,7 +165,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Null for no address provided"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def req = BlockchainOuterClass.BalanceRequest.newBuilder() .build() when: @@ -182,7 +176,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Build proto for common balance"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def balance = new TrackBitcoinAddress.AddressBalance(Chain.BITCOIN, "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK", BigInteger.valueOf(123456)) when: def act = track.buildResponse(balance) @@ -195,7 +189,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Build proto for zero balance"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def balance = new TrackBitcoinAddress.AddressBalance(Chain.BITCOIN, "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK", BigInteger.ZERO) when: def act = track.buildResponse(balance) @@ -208,7 +202,7 @@ class TrackBitcoinAddressSpec extends Specification { def "Build proto for all bitcoins"() { setup: - TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(Upstreams)) + TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) def balance = new TrackBitcoinAddress.AddressBalance(Chain.BITCOIN, "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK", BigInteger.valueOf(21_000_000).multiply(BigInteger.TEN.pow(8))) when: def act = track.buildResponse(balance) @@ -227,7 +221,7 @@ class TrackBitcoinAddressSpec extends Specification { 1 * getFlux() >> Flux.from(blocks) } def upstream = null - upstream = Mock(BitcoinChainUpstreams) { + upstream = Mock(BitcoinMultistream) { _ * getReader() >> Mock(BitcoinReader) { 2 * listUnspent() >>> [ Mono.just([]), @@ -239,7 +233,7 @@ class TrackBitcoinAddressSpec extends Specification { upstream } } - Upstreams upstreams = new UpstreamsMock(Chain.BITCOIN, upstream) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.BITCOIN, upstream) TrackBitcoinAddress track = new TrackBitcoinAddress(upstreams) when: diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy index ed104b22..4a04dee5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy @@ -17,13 +17,10 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinReader -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.bitcoin.CachingMempoolData import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux @@ -38,7 +35,7 @@ class TrackBitcoinTxSpec extends Specification { def "loadMempool() returns not found when not found"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) CachingMempoolData mempoolAccess = Mock(CachingMempoolData) { 1 * get() >> Mono.just([ @@ -46,7 +43,7 @@ class TrackBitcoinTxSpec extends Specification { "d296c6d47335a7f283574b06f1d6303b30ac75631e081ab128346a549ad93350" ]) } - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { _ * getReader() >> Mock(BitcoinReader) { _ * getMempool() >> mempoolAccess } @@ -65,14 +62,14 @@ class TrackBitcoinTxSpec extends Specification { def "loadMempool() returns ok when found"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) CachingMempoolData mempoolAccess = Mock(CachingMempoolData) { 1 * get() >> Mono.just([ "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9", "d296c6d47335a7f283574b06f1d6303b30ac75631e081ab128346a549ad93350" ]) } - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { _ * getReader() >> Mock(BitcoinReader) { _ * getMempool() >> mempoolAccess } @@ -91,9 +88,9 @@ class TrackBitcoinTxSpec extends Specification { def "loadExiting() returns not found if not mined"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { _ * getReader() >> Mock(BitcoinReader) { 1 * getTx(txid) >> Mono.just([ txid: txid @@ -114,9 +111,9 @@ class TrackBitcoinTxSpec extends Specification { def "loadExiting() returns block if mined"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { _ * getReader() >> Mock(BitcoinReader) { 1 * getTx(txid) >> Mono.just([ txid : txid, @@ -141,7 +138,7 @@ class TrackBitcoinTxSpec extends Specification { def "Goes with confirmations"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" // start with the current block def next = Flux.fromIterable([10, 12, 13, 14, 15]).map { h -> @@ -150,7 +147,7 @@ class TrackBitcoinTxSpec extends Specification { Head head = Mock(Head) { 1 * getFlux() >> next } - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { 1 * getHead() >> head } def status = new TrackBitcoinTx.TxStatus( @@ -172,7 +169,7 @@ class TrackBitcoinTxSpec extends Specification { def "Wait until mined"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" // start with the current block def next = Flux.fromIterable([10, 12, 13]).map { h -> @@ -196,7 +193,7 @@ class TrackBitcoinTxSpec extends Specification { ]) ] } - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { 1 * getHead() >> head _ * getReader() >> api } @@ -215,7 +212,7 @@ class TrackBitcoinTxSpec extends Specification { def "Check mempool until found"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" Head head = Mock(Head) { @@ -235,7 +232,7 @@ class TrackBitcoinTxSpec extends Specification { ]) _ * getMempool() >> mempoolAccess } - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { _ * getHead() >> head _ * getReader() >> api } @@ -256,7 +253,7 @@ class TrackBitcoinTxSpec extends Specification { def "Subscribe to an existing tx"() { setup: - TrackBitcoinTx track = new TrackBitcoinTx(Stub(Upstreams)) + TrackBitcoinTx track = new TrackBitcoinTx(Stub(MultistreamHolder)) def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" BitcoinReader api = Mock(BitcoinReader) { _ * getTx(txid) >> Mono.just([ @@ -276,7 +273,7 @@ class TrackBitcoinTxSpec extends Specification { Head head = Mock(Head) { _ * getFlux() >> next } - BitcoinChainUpstreams upstream = Mock(BitcoinChainUpstreams) { + BitcoinMultistream upstream = Mock(BitcoinMultistream) { _ * getReader() >> api _ * getHead() >> head } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy index e68f7843..71f74b96 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy @@ -19,21 +19,12 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.test.UpstreamsMock -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumReader +import io.emeraldpay.dshackle.test.MultistreamHolderMock +import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.BlockHash -import io.infinitape.etherjar.rpc.ReactorRpcClient -import io.infinitape.etherjar.rpc.RpcCall import io.infinitape.etherjar.rpc.json.BlockJson -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor -import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -66,7 +57,7 @@ class TrackEthereumAddressSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumAddress trackAddress = new TrackEthereumAddress(upstreams) apiMock.answer("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") @@ -106,7 +97,7 @@ class TrackEthereumAddressSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumAddress trackAddress = new TrackEthereumAddress(upstreams) apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index d99af948..72d4a60c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy @@ -23,10 +23,10 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.test.UpstreamsMock +import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -98,7 +98,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) apiMock.answer("eth_getTransactionByHash", [txId], txJson) @@ -118,8 +118,8 @@ class TrackEthereumTxSpec extends Specification { setup: def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - ((EthereumChainUpstream) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) { + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) + ((EthereumMultistream) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) { _ * getFlux() >> Flux.empty() } TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) @@ -131,7 +131,7 @@ class TrackEthereumTxSpec extends Specification { when: def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def act = StepVerifier.withVirtualTime( - { trackTx.subscribe(tx, upstreams.getUpstream(Chain.ETHEREUM).cast(EthereumChainUpstream)) }, + { trackTx.subscribe(tx, upstreams.getUpstream(Chain.ETHEREUM).cast(EthereumMultistream)) }, { scheduler }, 5) @@ -168,7 +168,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) def scheduler = VirtualTimeScheduler.create(true) trackTx.scheduler = scheduler @@ -193,7 +193,7 @@ class TrackEthereumTxSpec extends Specification { setup: def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) @@ -215,7 +215,7 @@ class TrackEthereumTxSpec extends Specification { setup: def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) @@ -288,7 +288,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) - Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) + MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) apiMock.answerOnce("eth_getTransactionByHash", [txId], null) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy similarity index 72% rename from src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy rename to src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 6876fbc2..6c8de274 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -16,47 +16,47 @@ */ package io.emeraldpay.dshackle.test -import com.fasterxml.jackson.databind.ObjectMapper + import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.upstream.AggregatedUpstream -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams +import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumReader import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux -class UpstreamsMock implements Upstreams { +class MultistreamHolderMock implements MultistreamHolder { private Map target = [:] - private Map upstreams = [:] + private Map upstreams = [:] - UpstreamsMock(Chain chain, Upstream up) { + MultistreamHolderMock(Chain chain, Upstream up) { addUpstream(chain, up) } - AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { + Multistream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { if (!upstreams.containsKey(chain)) { if (BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM) { - if (up instanceof EthereumChainUpstream) { + if (up instanceof EthereumMultistream) { upstreams[chain] = up } else if (up instanceof EthereumUpstream) { - upstreams[chain] = new EthereumChainUpstreamMock(chain, [up as EthereumUpstream], Caches.default(TestingCommons.objectMapper())) + upstreams[chain] = new EthereumMultistreamMock(chain, [up as EthereumUpstream], Caches.default(TestingCommons.objectMapper())) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } upstreams[chain].start() } else if (BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN) { - if (up instanceof BitcoinChainUpstreams) { + if (up instanceof BitcoinMultistream) { upstreams[chain] = up } else if (up instanceof BitcoinUpstream) { - upstreams[chain] = new BitcoinChainUpstreams(chain, [up as BitcoinUpstream], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) + upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinUpstream], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } @@ -69,7 +69,7 @@ class UpstreamsMock implements Upstreams { } @Override - AggregatedUpstream getUpstream(@NotNull Chain chain) { + Multistream getUpstream(@NotNull Chain chain) { return upstreams[chain] } @@ -97,11 +97,11 @@ class UpstreamsMock implements Upstreams { return upstreams.containsKey(chain) } - static class EthereumChainUpstreamMock extends EthereumChainUpstream { + static class EthereumMultistreamMock extends EthereumMultistream { EthereumReader customReader = null - EthereumChainUpstreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { + EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { super(chain, upstreams, caches, TestingCommons.objectMapper()) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index e81d904e..e46b3e9f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -25,9 +25,9 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.AggregatedUpstream +import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -71,12 +71,12 @@ class TestingCommons { return new EthereumUpstreamMock(Chain.ETHEREUM, api, new DirectCallMethods(methods)) } - static AggregatedUpstream aggregatedUpstream(Reader api) { + static Multistream aggregatedUpstream(Reader api) { return aggregatedUpstream(upstream(api)) } - static AggregatedUpstream aggregatedUpstream(EthereumUpstream up) { - return new EthereumChainUpstream(Chain.ETHEREUM, [up], Caches.default(objectMapper()), objectMapper()) + static Multistream aggregatedUpstream(EthereumUpstream up) { + return new EthereumMultistream(Chain.ETHEREUM, [up], Caches.default(objectMapper()), objectMapper()) } static CachesFactory emptyCaches() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy similarity index 87% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy index e0371811..3d320d7b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy @@ -19,14 +19,13 @@ import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.ReactorRpcClient import spock.lang.Specification -class CurrentUpstreamsSpec extends Specification { +class CurrentMultistreamHolderSpec extends Specification { def "add upstream"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api()) when: current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED)) @@ -37,7 +36,7 @@ class CurrentUpstreamsSpec extends Specification { def "add multiple upstreams"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) @@ -53,7 +52,7 @@ class CurrentUpstreamsSpec extends Specification { def "remove upstream"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) @@ -71,7 +70,7 @@ class CurrentUpstreamsSpec extends Specification { def "available after adding"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) + def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) when: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy similarity index 86% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index e8d73944..aa243140 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -21,17 +21,17 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.grpc.Chain import spock.lang.Specification -class AggregatedUpstreamSpec extends Specification { +class MultistreamSpec extends Specification { def "Aggregates methods"() { setup: def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new EthereumChainUpstream(Chain.ETHEREUM, [up1, up2], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) + def aggr = new EthereumMultistream(Chain.ETHEREUM, [up1, up2], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy index 602af14e..c8ee364c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy @@ -17,26 +17,18 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.cache.BlocksMemCache import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.cache.HeightCache import io.emeraldpay.dshackle.cache.TxMemCache import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxContainer -import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.test.UpstreamsMock -import io.emeraldpay.dshackle.upstream.AggregatedUpstream -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.Wei -import io.infinitape.etherjar.rpc.ReactorRpcClient -import io.infinitape.etherjar.rpc.RpcException -import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionJson import io.infinitape.etherjar.rpc.json.TransactionRefJson @@ -44,7 +36,6 @@ import reactor.core.publisher.Mono import spock.lang.Specification import java.time.Instant -import java.time.temporal.ChronoUnit class EthereumReaderSpec extends Specification { @@ -73,7 +64,7 @@ class EthereumReaderSpec extends Specification { .setBlockByHash(memCache) .setObjectMapper(TestingCommons.objectMapper()) .build() - def reader = new EthereumReader(Stub(AggregatedUpstream), caches, TestingCommons.objectMapper()) + def reader = new EthereumReader(Stub(Multistream), caches, TestingCommons.objectMapper()) when: def act = reader.blocksById().read(blockId).block() @@ -135,7 +126,7 @@ class EthereumReaderSpec extends Specification { .setBlockByHash(memCache) .setObjectMapper(TestingCommons.objectMapper()) .build() - def reader = new EthereumReader(Stub(AggregatedUpstream), caches, TestingCommons.objectMapper()) + def reader = new EthereumReader(Stub(Multistream), caches, TestingCommons.objectMapper()) when: def act = reader.blocksByHash().read(blockJson.hash).block() @@ -174,7 +165,7 @@ class EthereumReaderSpec extends Specification { .setTxByHash(memCache) .setObjectMapper(TestingCommons.objectMapper()) .build() - def reader = new EthereumReader(Stub(AggregatedUpstream), caches, TestingCommons.objectMapper()) + def reader = new EthereumReader(Stub(Multistream), caches, TestingCommons.objectMapper()) when: def act = reader.txByHash().read(txJson.hash).block() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index 54b8fd5a..5b236dde 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -29,8 +29,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.BlockHash -import io.infinitape.etherjar.rpc.ReactorRpcClient -import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import io.infinitape.etherjar.rpc.json.BlockJson import spock.lang.Specification