From 0ab68ea7827c609ae5fe7a7f8766faa4540d89de Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 11 Apr 2020 22:55:47 -0400 Subject: [PATCH] solution: refactor upstreams to make them blockchain specific --- .../io/emeraldpay/dshackle/BlockchainType.kt | 41 ++++++ .../dshackle/config/UpstreamsConfig.kt | 9 ++ .../dshackle/quorum/AlwaysQuorum.kt | 4 +- .../dshackle/quorum/BroadcastQuorum.kt | 4 +- .../emeraldpay/dshackle/quorum/CallQuorum.kt | 8 +- .../dshackle/quorum/NonEmptyQuorum.kt | 6 +- .../emeraldpay/dshackle/quorum/NonceQuorum.kt | 6 +- .../dshackle/quorum/NotLaggingQuorum.kt | 4 +- .../dshackle/quorum/ValueAwareQuorum.kt | 8 +- .../dshackle/reader/BlockApiReader.kt | 44 ------ .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 13 +- .../io/emeraldpay/dshackle/rpc/Describe.kt | 4 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 16 ++- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 20 ++- .../dshackle/rpc/SubscribeStatus.kt | 4 +- ...rackAddress.kt => TrackEthereumAddress.kt} | 19 ++- .../rpc/{TrackTx.kt => TrackEthereumTx.kt} | 18 ++- .../dshackle/startup/ConfiguredUpstreams.kt | 18 +-- .../dshackle/startup/UpstreamChange.kt | 27 +++- .../dshackle/upstream/AggregatedUpstream.kt | 51 +++---- .../emeraldpay/dshackle/upstream/ApiSource.kt | 2 +- .../dshackle/upstream/ChainUpstreams.kt | 99 +++----------- .../dshackle/upstream/CurrentUpstreams.kt | 23 ++-- .../dshackle/upstream/DefaultUpstream.kt | 4 +- .../dshackle/upstream/FilteredApis.kt | 23 ++-- .../io/emeraldpay/dshackle/upstream/Head.kt | 4 +- .../dshackle/upstream/HeadLagObserver.kt | 43 +++--- .../emeraldpay/dshackle/upstream/Selector.kt | 13 +- .../emeraldpay/dshackle/upstream/Upstream.kt | 8 +- .../dshackle/upstream/UpstreamApi.kt | 32 +++++ .../emeraldpay/dshackle/upstream/Upstreams.kt | 2 +- ...edMethods.kt => DefaultEthereumMethods.kt} | 3 +- .../dshackle/upstream/ethereum/EthereumApi.kt | 9 +- .../ethereum/EthereumChainUpstreams.kt | 127 ++++++++++++++++++ .../ethereum/EthereumHeadLagObserver.kt | 47 +++++++ .../upstream/ethereum/EthereumUpstream.kt | 20 ++- ...rpcUpstream.kt => EthereumGrpcUpstream.kt} | 21 ++- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 13 +- .../dshackle/BlockchainTypeSpec.groovy | 22 +++ .../dshackle/rpc/StreamHeadSpec.groovy | 3 +- ...groovy => TrackEthereumAddressSpec.groovy} | 6 +- ...Spec.groovy => TrackEthereumTxSpec.groovy} | 18 +-- .../dshackle/test/EthereumUpstreamMock.groovy | 6 +- .../dshackle/test/TestingCommons.groovy | 3 +- .../dshackle/test/UpstreamsMock.groovy | 11 +- .../upstream/AggregatedUpstreamSpec.groovy | 3 +- .../dshackle/upstream/FilteredApisSpec.groovy | 4 +- .../EthereumHeadLagObserverSpec.groovy} | 13 +- ...groovy => EthereumGrpcUpstreamSpec.groovy} | 8 +- 49 files changed, 600 insertions(+), 314 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt rename src/main/kotlin/io/emeraldpay/dshackle/rpc/{TrackAddress.kt => TrackEthereumAddress.kt} (89%) rename src/main/kotlin/io/emeraldpay/dshackle/rpc/{TrackTx.kt => TrackEthereumTx.kt} (94%) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamApi.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/{QuorumBasedMethods.kt => DefaultEthereumMethods.kt} (99%) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/{GrpcUpstream.kt => EthereumGrpcUpstream.kt} (89%) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/BlockchainTypeSpec.groovy rename src/test/groovy/io/emeraldpay/dshackle/rpc/{TrackAddressSpec.groovy => TrackEthereumAddressSpec.groovy} (94%) rename src/test/groovy/io/emeraldpay/dshackle/rpc/{TrackTxSpec.groovy => TrackEthereumTxSpec.groovy} (95%) rename src/test/groovy/io/emeraldpay/dshackle/upstream/{HeadLagObserverSpec.groovy => ethereum/EthereumHeadLagObserverSpec.groovy} (89%) rename src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/{GrpcUpstreamSpec.groovy => EthereumGrpcUpstreamSpec.groovy} (96%) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt b/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt new file mode 100644 index 00000000..9080149a --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/BlockchainType.kt @@ -0,0 +1,41 @@ +/** + * Copyright (c) 2020 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 + +import io.emeraldpay.grpc.Chain + +/** + * Type of the blockchain, architecture-wise + */ +enum class BlockchainType { + + ETHEREUM, + BITCOIN, + OTHER; + + companion object { + @JvmStatic + fun fromBlockchain(blockchain: Chain): BlockchainType { + if (blockchain == Chain.ETHEREUM || blockchain == Chain.ETHEREUM_CLASSIC || blockchain == Chain.TESTNET_KOVAN || blockchain == Chain.TESTNET_MORDEN) { + return ETHEREUM + } + if (blockchain == Chain.BITCOIN || blockchain == Chain.TESTNET_BITCOIN) { + return BITCOIN + } + return OTHER + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 5d1e6c83..40db6ea4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.config import io.emeraldpay.dshackle.Defaults +import java.lang.ClassCastException import java.net.URI import java.util.* import kotlin.collections.ArrayList @@ -72,6 +73,14 @@ class UpstreamsConfig { var connection: T? = null val labels = Labels() var methods: Methods? = null + + @Suppress("unchecked") + fun cast(type: Class): Upstream { + if (connection == null || type.isAssignableFrom(connection!!.javaClass)) { + return this as Upstream + } + throw ClassCastException("Cannot cast ${connection?.javaClass} to $type") + } } open class UpstreamConnection diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index f54e06fc..34920888 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -34,13 +34,13 @@ open class AlwaysQuorum: CallQuorum { return resolved } - override fun record(response: ByteArray, upstream: Upstream): Boolean { + override fun record(response: ByteArray, upstream: Upstream<*, *>): Boolean { result = response resolved = true return true } - override fun record(error: RpcException, upstream: Upstream) { + override fun record(error: RpcException, upstream: Upstream<*, *>) { } override fun getResult(): ByteArray? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt index 026d76cb..b33565ba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt @@ -42,7 +42,7 @@ open class BroadcastQuorum( return result } - override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) { + override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream<*, *>) { calls++ if (txid == null && responseValue != null) { txid = responseValue @@ -50,7 +50,7 @@ open class BroadcastQuorum( } } - override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*, *>) { // can be "message: known transaction: TXID", "Transaction with the same hash was already imported" or "message: Nonce too low" calls++ if (result == null) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index 018d3016..ac20594b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -30,8 +30,8 @@ interface CallQuorum { fun init(head: Head>) fun isResolved(): Boolean - fun record(response: ByteArray, upstream: Upstream): Boolean - fun record(error: RpcException, upstream: Upstream) + fun record(response: ByteArray, upstream: Upstream<*, *>): Boolean + fun record(error: RpcException, upstream: Upstream<*, *>) fun getResult(): ByteArray? companion object { @@ -41,8 +41,8 @@ interface CallQuorum { } } - fun asReducer(): BiFunction, CallQuorum> { - return BiFunction, CallQuorum> { a, b -> + fun asReducer(): BiFunction>, CallQuorum> { + return BiFunction>, CallQuorum> { a, b -> a.record(b.t1, b.t2) return@BiFunction a } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt index d801be5f..7f064ce7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt @@ -38,7 +38,7 @@ open class NonEmptyQuorum( return result != null || tries >= maxTries } - override fun recordValue(response: ByteArray, responseValue: Any?, upstream: Upstream) { + override fun recordValue(response: ByteArray, responseValue: Any?, upstream: Upstream<*, *>) { tries++ if (responseValue != null) { result = response @@ -49,10 +49,10 @@ open class NonEmptyQuorum( return result } - override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*, *>) { } - override fun record(error: RpcException, upstream: Upstream) { + override fun record(error: RpcException, upstream: Upstream<*, *>) { } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt index d41b5d17..954bae99 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt @@ -46,7 +46,7 @@ open class NonceQuorum( } } - override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) { + override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream<*, *>) { val value = responseValue?.let { str -> HexQuantity.from(str).value.toLong() } @@ -65,11 +65,11 @@ open class NonceQuorum( return result } - override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*, *>) { errors++ } - override fun record(error: RpcException, upstream: Upstream) { + override fun record(error: RpcException, upstream: Upstream<*, *>) { errors++ } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index 10e82e53..cc3405e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -34,7 +34,7 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { return result.get() != null } - override fun record(response: ByteArray, upstream: Upstream): Boolean { + override fun record(response: ByteArray, upstream: Upstream<*, *>): Boolean { val lagging = upstream.getLag() > maxLag if (!lagging) { result.set(response) @@ -43,7 +43,7 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { return false } - override fun record(error: RpcException, upstream: Upstream) { + override fun record(error: RpcException, upstream: Upstream<*, *>) { } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt index ffc8c8f8..9ae8cab9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt @@ -31,7 +31,7 @@ abstract class ValueAwareQuorum( return jacksonRpcConverter.fromJson(response.inputStream(), clazz) } - override fun record(response: ByteArray, upstream: Upstream): Boolean { + override fun record(response: ByteArray, upstream: Upstream<*, *>): Boolean { try { val value = extractValue(response, clazz) recordValue(response, value, upstream) @@ -43,12 +43,12 @@ abstract class ValueAwareQuorum( return isResolved(); } - override fun record(error: RpcException, upstream: Upstream) { + override fun record(error: RpcException, upstream: Upstream<*, *>) { recordError(null, error.rpcMessage, upstream) } - abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream) + abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream<*, *>) - abstract fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) + abstract fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*, *>) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt deleted file mode 100644 index ab7cd650..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt +++ /dev/null @@ -1,44 +0,0 @@ -/** - * 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.reader - -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream -import io.infinitape.etherjar.domain.BlockHash -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.Commands -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson -import reactor.core.publisher.Mono -import reactor.retry.Repeat -import java.time.Duration - -class BlockApiReader( - val upstream: Upstream -): Reader> { - - override fun read(key: BlockHash): Mono> { - return Mono.just(key) - .flatMap { - upstream.getApi(Selector.empty) - .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it)) } - }.repeatWhenEmpty { n -> - Repeat.times(3) - .exponentialBackoff(Duration.ofMillis(100), Duration.ofMillis(500)) - .apply(n) - } - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 8e14c3d3..20edca5d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -18,7 +18,8 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc -import io.grpc.stub.StreamObserver +import io.emeraldpay.dshackle.BlockchainType +import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -29,8 +30,8 @@ import reactor.core.publisher.Mono class BlockchainRpc( @Autowired private val nativeCall: NativeCall, @Autowired private val streamHead: StreamHead, - @Autowired private val trackTx: TrackTx, - @Autowired private val trackAddress: TrackAddress, + @Autowired private val trackEthereumTx: TrackEthereumTx, + @Autowired private val trackEthereumAddress: TrackEthereumAddress, @Autowired private val describe: Describe, @Autowired private val subscribeStatus: SubscribeStatus ): ReactorBlockchainGrpc.BlockchainImplBase() { @@ -46,15 +47,15 @@ class BlockchainRpc( } override fun subscribeTxStatus(request: Mono): Flux { - return trackTx.add(request) + return trackEthereumTx.add(request) } override fun subscribeBalance(request: Mono): Flux { - return trackAddress.subscribe(request) + return trackEthereumAddress.subscribe(request) } override fun getBalance(request: Mono): Flux { - return trackAddress.getBalance(request) + return trackEthereumAddress.getBalance(request) } override fun describe(request: Mono): Mono { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 2ca6cb44..e0f20a6b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -20,7 +20,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream -import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream +import io.emeraldpay.dshackle.upstream.grpc.EthereumGrpcUpstream import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.core.publisher.Mono @@ -47,7 +47,7 @@ class Describe( val nodes = QuorumForLabels() if (up is EthereumUpstream) { nodes.add(up.node) - } else if (up is GrpcUpstream) { + } else if (up is EthereumGrpcUpstream) { nodes.add(up.getNodes()) } nodes.getAll().forEach { node -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index e94e61b1..fc8baef3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -18,12 +18,16 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum +import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -84,13 +88,17 @@ open class NativeCall( if (chain == Chain.UNSPECIFIED) { return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) } + if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) { + return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number))) + } + val upstream = upstreams.getUpstream(chain) ?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain))) - return prepareCall(request, upstream) + return prepareCall(request, upstream as AggregatedUpstream>) } - fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux> { + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream>): Flux> { return request.itemsList.toFlux().map { val method = it.method val params = it.payload.toStringUtf8() @@ -197,7 +205,7 @@ open class NativeCall( } open class CallContext(val id: Int, - val upstream: AggregatedUpstream, + val upstream: AggregatedUpstream>, val matcher: Selector.Matcher, val callQuorum: CallQuorum, val payload: T) { @@ -205,7 +213,7 @@ open class NativeCall( return CallContext(id, upstream, matcher, callQuorum, payload) } - fun getApis(): ApiSource { + fun getApis(): ApiSource { return upstream.getApis(matcher) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 3ab3b554..c6c289fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -18,9 +18,9 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory @@ -44,11 +44,25 @@ class StreamHead( ?: return@flatMapMany Flux.error(Exception("Unavailable chain: $chain")) up.getHead() .getFlux() - .map { asProto(chain, it) } + .map { asProto(chain, it!!) } + .onErrorContinue { t, _ -> + log.warn("Head error: ${t.message}") + } } } - fun asProto(chain: Chain, block: BlockJson): BlockchainOuterClass.ChainHead { + fun asProto(chain: Chain, block: Any): BlockchainOuterClass.ChainHead { + if (BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM) { + if (BlockJson::class.java.isAssignableFrom(block.javaClass)) { + return asEthereumProto(chain, block as BlockJson) + } else { + throw IllegalArgumentException("Invalid block type: ${block.javaClass}") + } + } + throw IllegalArgumentException("Unsupported blockchain ${chain}") + } + + fun asEthereumProto(chain: Chain, block: BlockJson): BlockchainOuterClass.ChainHead { return BlockchainOuterClass.ChainHead.newBuilder() .setChainValue(chain.id) .setHeight(block.number) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index f8c14c1d..5fc42eee 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -45,7 +45,7 @@ class SubscribeStatus( } } - fun chainStatus(chain: Chain, ups: List): BlockchainOuterClass.ChainStatus { + fun chainStatus(chain: Chain, ups: List>): BlockchainOuterClass.ChainStatus { val available = ups.map { u -> u.getStatus() }.min() ?: UpstreamAvailability.UNAVAILABLE @@ -59,6 +59,6 @@ class SubscribeStatus( .build() } - class ChainSubscription(val chain: Chain, val up: AggregatedUpstream, val avail: UpstreamAvailability) + class ChainSubscription(val chain: Chain, val up: AggregatedUpstream<*, *>, val avail: UpstreamAvailability) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt similarity index 89% rename from src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt rename to src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt index c23192d1..555501b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt @@ -17,15 +17,20 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass 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.AggregatedUpstream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.Wei import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockTag +import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.scheduling.annotation.Scheduled @@ -43,12 +48,12 @@ import java.util.concurrent.atomic.AtomicLong import javax.annotation.PostConstruct @Service -class TrackAddress( +class TrackEthereumAddress( @Autowired private val upstreams: Upstreams, @Autowired private val upstreamScheduler: Scheduler ) { - private val log = LoggerFactory.getLogger(TrackAddress::class.java) + private val log = LoggerFactory.getLogger(TrackEthereumAddress::class.java) private val clients = HashMap>() private val seq = AtomicLong(0) @@ -94,6 +99,9 @@ class TrackAddress( private fun initializeSimple(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) + if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) { + return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) + } if (!upstreams.isAvailable(chain)) { return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) } @@ -122,6 +130,10 @@ class TrackAddress( fun subscribe(requestMono: Mono): Flux { return requestMono.flatMapMany { request -> + val chain = Chain.byId(request.asset.chainValue) + if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) { + return@flatMapMany Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) + } val bus = TopicProcessor.create() initializeSubscription(request, bus) .flatMap { tracked -> @@ -175,7 +187,8 @@ class TrackAddress( } fun getBalance(addr: SimpleAddress): Mono { - val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(SilentException.UnsupportedBlockchain(addr.chain)) + val up = upstreams.getUpstream(addr.chain) as AggregatedUpstream>? + ?: return Mono.error(SilentException.UnsupportedBlockchain(addr.chain)) return up.getApi(Selector.empty) .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) } .timeout(Defaults.timeout) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt similarity index 94% rename from src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt rename to src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt index fda779c6..ac2b9981 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt @@ -18,10 +18,13 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString 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.AggregatedUpstream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -40,7 +43,6 @@ import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import reactor.core.scheduler.Scheduler import reactor.util.function.Tuples -import java.lang.Exception import java.math.BigInteger import java.time.Duration import java.time.Instant @@ -53,7 +55,7 @@ import kotlin.math.max import kotlin.math.min @Service -class TrackTx( +class TrackEthereumTx( @Autowired private val upstreams: Upstreams, @Autowired private val upstreamScheduler: Scheduler ) { @@ -67,7 +69,7 @@ class TrackTx( private val PING_PERIOD = Duration.ofMinutes(5) } - private val log = LoggerFactory.getLogger(TrackTx::class.java) + private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java) private val clients = HashMap>() private val seq = AtomicLong(0) @@ -142,6 +144,9 @@ class TrackTx( fun prepareTracking(request: BlockchainOuterClass.TxStatusRequest): TxDetails { val chain = Chain.byId(request.chainValue) + if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) { + throw SilentException.UnsupportedBlockchain(request.chainValue) + } if (!clients.containsKey(chain)) { throw SilentException.UnsupportedBlockchain(chain) } @@ -208,7 +213,7 @@ class TrackTx( } private fun loadWeight(tx: TxDetails): Mono { - val upstream = upstreams.getUpstream(tx.chain) + val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream>? ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain)) return upstream.getApi(Selector.empty) .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) } @@ -219,7 +224,7 @@ class TrackTx( } } - fun updateFromBlock(upstream: Upstream, tx: TxDetails, it: TransactionJson): Mono { + fun updateFromBlock(upstream: Upstream>, tx: TxDetails, it: TransactionJson): Mono { return if (it.blockNumber != null && it.blockHash != null && it.blockHash != ZERO_BLOCK) { val updated = tx.withStatus( blockHash = it.blockHash, @@ -250,7 +255,8 @@ class TrackTx( private fun checkForUpdate(tx: TxDetails): Mono { val initialStatus = tx.status - val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain)) + val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream>? + ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain)) val execution = upstream.getApi(Selector.empty) .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) } return execution diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 29829d27..e159053c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -16,9 +16,9 @@ package io.emeraldpay.dshackle.startup import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.config.UpstreamsConfigReader import io.emeraldpay.dshackle.upstream.CurrentUpstreams import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi @@ -27,16 +27,13 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.http.ReactorHttpRpcClient -import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired -import org.springframework.core.env.Environment import org.springframework.stereotype.Repository import java.net.URI import java.util.* import javax.annotation.PostConstruct import kotlin.collections.HashMap -import kotlin.system.exitProcess @Repository open class ConfiguredUpstreams( @@ -59,21 +56,26 @@ open class ConfiguredUpstreams( @PostConstruct fun start() { + log.debug("Starting upstreams") val defaultOptions = buildDefaultOptions(config) config.upstreams.forEach { up -> - + log.debug("Start upstream ${up.id}") if (up.connection is UpstreamsConfig.GrpcConnection) { val options = up.options ?: UpstreamsConfig.Options() - buildGrpcUpstream(up as UpstreamsConfig.Upstream, options) + buildGrpcUpstream(up.cast(UpstreamsConfig.GrpcConnection::class.java), options) } else { val chain = chainNames[up.chain] if (chain == null) { - log.error("Chain not supported: ${up.chain}") + log.error("Chain is unknown: ${up.chain}") + return@forEach + } + if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) { + log.error("Chain is unsupported: ${up.chain}") return@forEach } val options = (up.options ?: UpstreamsConfig.Options()) .merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults()) - buildEthereumUpstream(up as UpstreamsConfig.Upstream, chain, options) + buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt index d57061d7..c2535ae1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt @@ -20,15 +20,38 @@ import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.grpc.Chain +/** + * An update event to the list of currently available upstreams. + */ class UpstreamChange( + /** + * Target blockchain + */ val chain: Chain, - val upstream: Upstream, + /** + * Corresponding upstream + */ + val upstream: Upstream<*, *>, + /** + * Type of the change + */ val type: ChangeType ): CachesEnabled { + enum class ChangeType { + /** + * Upstream just added + */ ADDED, + + /** + * Upstream become available after being temporally off + */ REVALIDATED, - STALE, + + /** + * Upstream is removed (it still doesn't mean it wouldn't return again after some reconfiguration) + */ REMOVED, } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index 199f8fa1..a2061497 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -31,19 +31,22 @@ import java.util.concurrent.locks.ReentrantLock import java.util.function.Predicate import kotlin.concurrent.withLock -abstract class AggregatedUpstream( +/** + * Aggregation of multiple upstreams responding to a single blockchain + */ +abstract class AggregatedUpstream( private val objectMapper: ObjectMapper, val caches: Caches -): Upstream, Lifecycle { +) : Upstream, Lifecycle { private var cacheSubscription: Disposable? = null var cache: CachingEthereumApi = CachingEthereumApi.empty() private val reconfigLock = ReentrantLock() private var callMethods: CallMethods? = null - abstract fun getAll(): List - abstract fun addUpstream(upstream: Upstream) - abstract fun getApis(matcher: Selector.Matcher): ApiSource + abstract fun getAll(): List> + abstract fun addUpstream(upstream: Upstream) + abstract fun getApis(matcher: Selector.Matcher): ApiSource fun onUpstreamsUpdated() { reconfigLock.withLock { @@ -78,24 +81,6 @@ abstract class AggregatedUpstream( return callMethods ?: throw IllegalStateException("Methods are not initialized yet") } - class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) - - class FilterBestAvailability(): Predicate { - private val lastRef = AtomicReference() - - override fun test(t: UpstreamStatus): Boolean { - val last = lastRef.get() - val changed = last == null - || t.status > last.status - || (last.upstream == t.upstream && t.status != last.status) - || last.ts.isBefore(Instant.now() - Duration.ofSeconds(60)) - if (changed) { - lastRef.set(t) - } - return changed - } - } - override fun start() { } @@ -114,4 +99,24 @@ abstract class AggregatedUpstream( } } + // -------------------------------------------------------------------------------------------------------- + + class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) + + class FilterBestAvailability() : Predicate> { + private val lastRef = AtomicReference>() + + override fun test(t: UpstreamStatus<*>): Boolean { + val last = lastRef.get() + val changed = last == null + || t.status > last.status + || (last.upstream == t.upstream && t.status != last.status) + || last.ts.isBefore(Instant.now() - Duration.ofSeconds(60)) + if (changed) { + lastRef.set(t) + } + return changed + } + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt index 4024d2ac..1bc3ed3d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt @@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import org.reactivestreams.Publisher -interface ApiSource: Publisher { +interface ApiSource : Publisher { fun resolve() fun request(tries: Int) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index a9d67a62..75788f70 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -17,38 +17,34 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi -import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead -import io.emeraldpay.dshackle.upstream.ethereum.EthereumHeadMerge import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Mono -import java.lang.IllegalStateException -import java.time.Duration -open class ChainUpstreams ( +/** + * General interface to upstream(s) to a single chain + */ +abstract class ChainUpstreams( val chain: Chain, - private val upstreams: MutableList, + private val upstreams: MutableList>, caches: Caches, objectMapper: ObjectMapper -) : AggregatedUpstream(objectMapper, caches), Lifecycle { +) : AggregatedUpstream(objectMapper, caches), Lifecycle { private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 - private var head: EthereumHead? = null - private var lagObserver: HeadLagObserver? = null + protected var lagObserver: HeadLagObserver? = null private var subscription: Disposable? = null - init { - if (upstreams.size > 0) { - head = updateHead() - onUpstreamsUpdated() - } + open fun init() { + onUpstreamsUpdated() } + abstract fun updateHead(): Head + abstract fun setHead(head: Head) + override fun getId(): String { return "!all:${chain.chainCode}" } @@ -68,7 +64,7 @@ open class ChainUpstreams ( super.stop() subscription?.dispose() subscription = null - head?.let { + getHead().let { if (it is Lifecycle) { it.stop() } @@ -76,50 +72,24 @@ open class ChainUpstreams ( lagObserver?.stop() } - internal fun updateHead(): EthereumHead { - head?.let { - if (it is Lifecycle) { - it.stop() - } - } - lagObserver?.stop() - lagObserver = null - val head = if (upstreams.size == 1) { - val upstream = upstreams.first() - upstream.setLag(0) - upstream.getHead() - } else { - val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }).apply { - this.start() - } - val lagObserver = HeadLagObserver(newHead, upstreams).apply { - this.start() - } - this.lagObserver = lagObserver - newHead - } - onHeadUpdated(head) - return head - } - - override fun getAll(): List { + override fun getAll(): List> { return upstreams } - override fun addUpstream(upstream: Upstream) { + override fun addUpstream(upstream: Upstream) { upstreams.add(upstream) - head = updateHead() + setHead(updateHead()) onUpstreamsUpdated() } fun removeUpstream(id: String) { if (upstreams.removeIf { it.getId() == id }) { - head = updateHead() + setHead(updateHead()) onUpstreamsUpdated() } } - override fun getApis(matcher: Selector.Matcher): ApiSource { + override fun getApis(matcher: Selector.Matcher): ApiSource { val i = seq++ if (seq >= Int.MAX_VALUE / 2) { seq = 0 @@ -127,15 +97,11 @@ open class ChainUpstreams ( return FilteredApis(upstreams, matcher, i) } - override fun getApi(matcher: Selector.Matcher): Mono { + override fun getApi(matcher: Selector.Matcher): Mono { val apis = getApis(matcher) apis.request(1) return Mono.from(apis) - .switchIfEmpty(Mono.error(Exception("No API available"))) - } - - override fun getHead(): EthereumHead { - return head!! + .switchIfEmpty(Mono.error(Exception("No API available"))) } override fun setLag(lag: Long) { @@ -145,28 +111,5 @@ open class ChainUpstreams ( return 0 } - override fun getLabels(): Collection { - return upstreams.flatMap { it.getLabels() } - } - - fun printStatus() { - var height: Long? = null - try { - height = head!!.getFlux().next().block(Duration.ofSeconds(1))?.number - } catch (e: IllegalStateException) { - //timout - } catch (e: Exception) { - log.warn("Head processing error: ${e.javaClass} ${e.message}") - } - val statuses = upstreams.map { it.getStatus() } - .groupBy { it } - .map { "${it.key.name}/${it.value.size}" } - .joinToString(",") - val lag = upstreams.map { it.getLag() } - .joinToString(", ") - - log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]") - } - - + abstract fun printStatus() } \ 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/CurrentUpstreams.kt index 72eaac65..f88b5089 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt @@ -16,13 +16,17 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods +import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream 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.scheduling.annotation.Scheduled @@ -42,22 +46,23 @@ class CurrentUpstreams( private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java) - private val chainMapping = ConcurrentHashMap() + private val chainMapping = ConcurrentHashMap>() private val chainsBus = TopicProcessor.create() - private val callTargets = HashMap() + private val callTargets = HashMap() private val updateLock = ReentrantLock() fun update(change: UpstreamChange) { updateLock.withLock { val chain = change.chain val up = change.upstream - val current = chainMapping[chain] + .cast(EthereumUpstream::class.java, EthereumApi::class.java, BlockJson::class.java) as Upstream> + val current = chainMapping[chain] as ChainUpstreams>? if (change.type == UpstreamChange.ChangeType.REMOVED) { current?.removeUpstream(up.getId()) log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed") } else { if (current == null) { - val created = ChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) + val created = EthereumChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) if (up is CachesEnabled) { up.setCaches(created.caches) } @@ -79,7 +84,7 @@ class CurrentUpstreams( } } - override fun getUpstream(chain: Chain): AggregatedUpstream? { + override fun getUpstream(chain: Chain): AggregatedUpstream<*, *>? { return chainMapping[chain] } @@ -103,8 +108,8 @@ class CurrentUpstreams( return callTargets[chain] ?: return setupDefaultMethods(chain) } - fun setupDefaultMethods(chain: Chain): QuorumBasedMethods { - val created = QuorumBasedMethods(objectMapper, chain) + fun setupDefaultMethods(chain: Chain): CallMethods { + val created = DefaultEthereumMethods(objectMapper, chain) callTargets[chain] = created return created } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 80a81faa..9cfc6d26 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -19,10 +19,10 @@ import reactor.core.publisher.Flux import reactor.core.publisher.TopicProcessor import java.util.concurrent.atomic.AtomicReference -abstract class DefaultUpstream( +abstract class DefaultUpstream( defaultLag: Long, defaultAvail: UpstreamAvailability -) : Upstream { +) : Upstream { constructor() : this(Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index a0646115..11c577fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -15,7 +15,6 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import org.reactivestreams.Subscriber import reactor.core.publisher.EmitterProcessor import reactor.core.publisher.Flux @@ -26,28 +25,28 @@ import kotlin.math.pow import kotlin.math.roundToLong import kotlin.random.Random -class FilteredApis( - allUpstreams: List, +class FilteredApis( + allUpstreams: List>, private val matcher: Selector.Matcher, pos: Int, private val repeatLimit: Long, jitter: Int -): ApiSource { +) : ApiSource { companion object { private const val DEFAULT_DELAY_STEP = 100 private const val MAX_WAIT_MILLIS = 5000L } - constructor(allUpstreams: List, + constructor(allUpstreams: List>, matcher: Selector.Matcher, - pos: Int): this(allUpstreams, matcher, pos, 10, 7) + pos: Int) : this(allUpstreams, matcher, pos, 10, 7) - constructor(allUpstreams: List, - matcher: Selector.Matcher): this(allUpstreams, matcher, 0, 10, 10) + constructor(allUpstreams: List>, + matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10) private val delay: Int - private val upstreams: List + private val upstreams: List> private val control = EmitterProcessor.create(32, false) @@ -75,18 +74,18 @@ class FilteredApis( return Duration.ofMillis(time) } - override fun subscribe(subscriber: Subscriber) { + override fun subscribe(subscriber: Subscriber) { val first = Flux.fromIterable(upstreams) val retries = (1 until repeatLimit).map { r -> Flux.fromIterable(upstreams).delaySubscription(waitDuration(r)) }.let { Flux.concat(it) } Flux.concat(first, retries) - .filter(Upstream::isAvailable) + .filter(Upstream::isAvailable) .filter(matcher::matches) .flatMap { it.getApi(matcher) } .zipWith(control).map { it.t1 } - .subscribe(subscriber) + .subscribe(subscriber as Subscriber) } override fun resolve() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index 24e5daed..8914c2ad 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -18,6 +18,6 @@ package io.emeraldpay.dshackle.upstream import reactor.core.publisher.Flux import reactor.core.publisher.Mono -interface Head { - fun getFlux(): Flux +interface Head { + fun getFlux(): Flux } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index 78fb3af0..c6ca3d6f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt @@ -15,23 +15,21 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux -import reactor.core.publisher.toFlux import reactor.util.function.Tuple2 import reactor.util.function.Tuples -import java.time.Duration -class HeadLagObserver ( - private val master: EthereumHead, - private val followers: Collection -): Lifecycle { +/** + * Observer group of upstreams and defined a distance in blocks (lag) between a leader (best height/difficulty) and + * other upstreams. + */ +abstract class HeadLagObserver( + private val master: Head, + private val followers: Collection> +) : Lifecycle { private val log = LoggerFactory.getLogger(HeadLagObserver::class.java) @@ -50,7 +48,7 @@ class HeadLagObserver ( current = null } - private fun subscription(): Flux { + fun subscription(): Flux { return master.getFlux() .flatMap(this::probeFollowers) .map { item -> @@ -58,38 +56,29 @@ class HeadLagObserver ( } } - fun probeFollowers(top: BlockJson): Flux> { - return followers.toFlux() + fun probeFollowers(top: B): Flux>> { + return Flux.fromIterable(followers) .parallel(followers.size) .flatMap { mapLagging(top, it, getCurrentBlocks(it)) } .sequential() .onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) } } - fun getCurrentBlocks(up: Upstream): Flux> { - val head = up.getHead() - return head.getFlux().take(Duration.ofSeconds(1)) - } + abstract fun getCurrentBlocks(up: Upstream): Flux - fun mapLagging(top: BlockJson, up: Upstream, blocks: Flux>): Flux> { + fun mapLagging(top: B, up: Upstream, blocks: Flux): Flux>> { return blocks .map { extractDistance(top, it) } - .takeUntil{ lag -> lag <= 0L } + .takeUntil { lag -> lag <= 0L } .map { Tuples.of(it, up) } .doOnError { t -> log.warn("Failed to find distance for $up", t) } } - fun extractDistance(top: BlockJson, curr: BlockJson): Long { - return when { - curr.number > top.number -> if (curr.totalDifficulty >= top.totalDifficulty) 0 else forkDistance(top, curr) - curr.number == top.number -> if (curr.totalDifficulty == top.totalDifficulty) 0 else forkDistance(top, curr) - else -> top.number - curr.number - } - } + abstract fun extractDistance(top: B, curr: B): Long - fun forkDistance(top: BlockJson, curr: BlockJson): Long { + fun forkDistance(top: B, curr: B): Long { //TODO look for common ancestor? though it may be a corruption return 6 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index 3e7ebee0..6b521a7e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -95,13 +95,13 @@ class Selector { } interface Matcher { - fun matches(up: Upstream): Boolean + fun matches(up: Upstream): Boolean } class MultiMatcher( private val matchers: Collection ): Matcher { - override fun matches(up: Upstream): Boolean { + override fun matches(up: Upstream): Boolean { return matchers.all { it.matches(up) } } @@ -113,21 +113,22 @@ class Selector { class MethodMatcher( val method: String ): Matcher { - override fun matches(up: Upstream): Boolean { + override fun matches(up: Upstream): Boolean { return up.getMethods().isAllowed(method) } } abstract class LabelSelectorMatcher: Matcher { - override fun matches(up: Upstream): Boolean { + override fun matches(up: Upstream): Boolean { return up.getLabels().any(this::matches) } + abstract fun matches(labels: UpstreamsConfig.Labels): Boolean abstract fun asProto(): BlockchainOuterClass.Selector? } class EmptyMatcher: Matcher { - override fun matches(up: Upstream): Boolean { + override fun matches(up: Upstream): Boolean { return true } } @@ -142,7 +143,7 @@ class Selector { return null } - override fun matches(up: Upstream): Boolean { + override fun matches(up: Upstream): Boolean { return true } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index f42c0002..33d213da 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -22,16 +22,18 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import reactor.core.publisher.Flux import reactor.core.publisher.Mono -interface Upstream { +interface Upstream { fun isAvailable(): Boolean fun getStatus(): UpstreamAvailability fun observeStatus(): Flux - fun getHead(): EthereumHead - fun getApi(matcher: Selector.Matcher): Mono + fun getHead(): Head + fun getApi(matcher: Selector.Matcher): Mono fun getOptions(): UpstreamsConfig.Options fun setLag(lag: Long) fun getLag(): Long fun getLabels(): Collection fun getMethods(): CallMethods fun getId(): String + + fun , TA : UpstreamApi, BA> cast(selfType: Class, upstreamType: Class, blockType: Class): T } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamApi.kt new file mode 100644 index 00000000..e4d95784 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamApi.kt @@ -0,0 +1,32 @@ +/** + * Copyright (c) 2020 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 reactor.core.publisher.Mono + +/** + * A general interface to make a request to an Upstream API + */ +interface UpstreamApi { + + /** + * @param id an internal uniq id, if multiple requests are made in batch + * @param method JSON RPC method name + * @param params JSON RPC parameters, must be serializable into a JSON array + */ + fun execute(id: Int, method: String, params: List): Mono + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index f9e89874..13db5657 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -20,7 +20,7 @@ import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux interface Upstreams { - fun getUpstream(chain: Chain): AggregatedUpstream? + fun getUpstream(chain: Chain): AggregatedUpstream<*, *>? fun getAvailable(): List fun observeChains(): Flux fun getDefaultMethods(chain: Chain): CallMethods diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/QuorumBasedMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt similarity index 99% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/QuorumBasedMethods.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt index 17146ed3..ba0db30a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/QuorumBasedMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt @@ -26,7 +26,7 @@ import java.util.* * Default configuration for Ethereum based RPC. Defines optimal Quorum strategies for different methods, and provides * hardcoded results for base methods, such as `net_version`, `web3_clientVersion` and similar */ -class QuorumBasedMethods( +class DefaultEthereumMethods( private val objectMapper: ObjectMapper, private val chain: Chain ) : CallMethods { @@ -102,6 +102,7 @@ class QuorumBasedMethods( override fun isAllowed(method: String): Boolean { return allowedMethods.contains(method) } + override fun isHardcoded(method: String): Boolean { return hardcodedMethods.contains(method) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt index 3d2c8f04..56c58a6e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt @@ -17,23 +17,24 @@ package io.emeraldpay.dshackle.upstream.ethereum import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamApi import io.infinitape.etherjar.rpc.* +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.io.InputStream abstract class EthereumApi( objectMapper: ObjectMapper -) { +) : UpstreamApi { companion object { private val log = LoggerFactory.getLogger(EthereumApi::class.java) } private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) - var upstream: Upstream? = null - - abstract fun execute(id: Int, method: String, params: List): Mono + var upstream: Upstream>? = null fun execute(rpcCall: RpcCall): Mono { return execute(0, rpcCall.method, rpcCall.params as List) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt new file mode 100644 index 00000000..db324c10 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt @@ -0,0 +1,127 @@ +/** + * Copyright (c) 2020 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.ethereum + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.upstream.* +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import java.lang.IllegalStateException +import java.time.Duration + +class EthereumChainUpstreams( + chain: Chain, + val upstreams: MutableList, + caches: Caches, + objectMapper: ObjectMapper +) : ChainUpstreams>(chain, upstreams as MutableList>>, caches, objectMapper) { + + companion object { + private val log = LoggerFactory.getLogger(EthereumChainUpstreams::class.java) + } + + private var head: EthereumHead? = null + + init { + this.init() + } + + override fun init() { + if (upstreams.size > 0) { + head = updateHead() + } + super.init() + } + + override fun getHead(): EthereumHead { + return head!! + } + + override fun setHead(head: Head>) { + this.head = head as EthereumHead + } + + override fun updateHead(): EthereumHead { + head?.let { + if (it is Lifecycle) { + it.stop() + } + } + lagObserver?.stop() + lagObserver = null + val head = if (upstreams.size == 1) { + val upstream = upstreams.first() + upstream.setLag(0) + upstream.getHead() + } else { + val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }).apply { + this.start() + } + val lagObserver = EthereumHeadLagObserver(newHead, upstreams).apply { + this.start() + } + this.lagObserver = lagObserver + newHead + } + onHeadUpdated(head) + return head + } + + override fun getLabels(): Collection { + return upstreams.flatMap { it.getLabels() } + } + + override fun printStatus() { + var height: Long? = null + try { + height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.number + } catch (e: IllegalStateException) { + //timout + } catch (e: Exception) { + log.warn("Head processing error: ${e.javaClass} ${e.message}") + } + val statuses = upstreams.map { it.getStatus() } + .groupBy { it } + .map { "${it.key.name}/${it.value.size}" } + .joinToString(",") + val lag = upstreams.map { it.getLag() } + .joinToString(", ") + + log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]") + } + + @SuppressWarnings("unchecked") + override fun , TA : UpstreamApi, BA> cast(selfType: Class, upstreamType: Class, blockType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + if (!upstreamType.isAssignableFrom(EthereumApi::class.java)) { + throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType") + } + if (!blockType.isAssignableFrom(BlockJson::class.java)) { + throw ClassCastException("Cannot cast ${BlockJson::class.java} to $blockType") + } + return this as T + } + + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt new file mode 100644 index 00000000..0664f1d4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt @@ -0,0 +1,47 @@ +/** + * Copyright (c) 2020 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.ethereum + +import io.emeraldpay.dshackle.upstream.HeadLagObserver +import io.emeraldpay.dshackle.upstream.Upstream +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import java.time.Duration + +class EthereumHeadLagObserver( + master: EthereumHead, + followers: Collection>> +) : HeadLagObserver>(master, followers) { + + companion object { + private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java) + } + + override fun getCurrentBlocks(up: Upstream>): Flux> { + val head = up.getHead() + return Flux.from(head.getFlux()).take(Duration.ofSeconds(1)) + } + + override fun extractDistance(top: BlockJson, curr: BlockJson): Long { + return when { + curr.number > top.number -> if (curr.totalDifficulty >= top.totalDifficulty) 0 else forkDistance(top, curr) + curr.number == top.number -> if (curr.totalDifficulty == top.totalDifficulty) 0 else forkDistance(top, curr) + else -> top.number - curr.number + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 7546b3d9..40ad8fd7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -23,6 +23,8 @@ import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods 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.context.Lifecycle import reactor.core.Disposable @@ -37,9 +39,9 @@ open class EthereumUpstream( private val options: UpstreamsConfig.Options, val node: QuorumForLabels.QuorumItem, private val targets: CallMethods -): DefaultUpstream(), CachesEnabled, Lifecycle { +) : DefaultUpstream>(), Upstream>, CachesEnabled, Lifecycle { - constructor(id: String, chain: Chain, api: DirectEthereumApi): this(id, chain, api, null, + constructor(id: String, chain: Chain, api: DirectEthereumApi) : this(id, chain, api, null, UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), DirectCallMethods()) @@ -133,4 +135,18 @@ open class EthereumUpstream( return targets } + @Suppress("unchecked") + override fun , TA : UpstreamApi, BA> cast(selfType: Class, upstreamType: Class, blockType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + if (!upstreamType.isAssignableFrom(EthereumApi::class.java)) { + throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType") + } + if (!blockType.isAssignableFrom(BlockJson::class.java)) { + throw ClassCastException("Cannot cast ${BlockJson::class.java} to $blockType") + } + return this as T + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt similarity index 89% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 6f19cef8..bc239863 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -30,6 +30,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash @@ -51,16 +52,16 @@ import java.util.concurrent.atomic.AtomicReference import java.util.function.Function import kotlin.collections.ArrayList -open class GrpcUpstream( +open class EthereumGrpcUpstream( private val parentId: String, private val chain: Chain, private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, private val rpcClient: ReactorEmeraldClient -): DefaultUpstream(), CachesEnabled, Lifecycle { +) : DefaultUpstream>(), CachesEnabled, Lifecycle { private var allLabels: Collection = ArrayList() - private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) + private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) private var caches: Caches? = null private val options = UpstreamsConfig.Options.getDefaults() @@ -214,4 +215,18 @@ open class GrpcUpstream( this.caches = caches } + @SuppressWarnings("unchecked") + override fun , TA : UpstreamApi, BA> cast(selfType: Class, upstreamType: Class, blockType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + if (!upstreamType.isAssignableFrom(EthereumApi::class.java)) { + throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $upstreamType") + } + if (!blockType.isAssignableFrom(BlockJson::class.java)) { + throw ClassCastException("Cannot cast ${BlockJson::class.java} to $blockType") + } + return this as T + } + } \ No newline at end of file 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 41824419..f6465e15 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -18,10 +18,10 @@ package io.emeraldpay.dshackle.upstream.grpc import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.api.proto.BlockchainOuterClass 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.config.AuthConfig -import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.grpc.Chain @@ -54,7 +54,7 @@ class GrpcUpstreams( var timeout = Defaults.timeout private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null - private val known = HashMap() + private val known = HashMap() private val lock = ReentrantLock() private var grpcTransport: ReactorEmeraldClient? = null @@ -117,7 +117,7 @@ class GrpcUpstreams( }.map { chainDetails -> val chain = Chain.byId(chainDetails.chain.number) val up = getOrCreate(chain) - (up.upstream as GrpcUpstream).init(chainDetails) + (up.upstream as EthereumGrpcUpstream).init(chainDetails) up } @@ -155,10 +155,13 @@ class GrpcUpstreams( } fun getOrCreate(chain: Chain): UpstreamChange { + if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) { + throw IllegalArgumentException("Unsupported blockchain: $chain") + } lock.withLock { val current = known[chain] return if (current == null) { - val created = GrpcUpstream(id, chain, client!!, objectMapper, grpcTransport!!.copyForChain(chain)) + val created = EthereumGrpcUpstream(id, chain, client!!, objectMapper, grpcTransport!!.copyForChain(chain)) created.timeout = this.timeout known[chain] = created created.start() @@ -169,7 +172,7 @@ class GrpcUpstreams( } } - fun get(chain: Chain): GrpcUpstream { + fun get(chain: Chain): EthereumGrpcUpstream { return known[chain]!! } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/BlockchainTypeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/BlockchainTypeSpec.groovy new file mode 100644 index 00000000..7c3c3255 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/BlockchainTypeSpec.groovy @@ -0,0 +1,22 @@ +package io.emeraldpay.dshackle + +import io.emeraldpay.grpc.Chain +import spock.lang.Specification + +class BlockchainTypeSpec extends Specification { + + def "Correct type for ethereum"() { + expect: + BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM + where: + chain << [Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_KOVAN, Chain.TESTNET_MORDEN] + } + + def "Correct type for bitcoin"() { + expect: + BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN + where: + chain << [Chain.BITCOIN, Chain.TESTNET_BITCOIN] + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy index 16286829..2280fd03 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.UpstreamsMock import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -38,7 +39,7 @@ class StreamHeadSpec extends Specification { def "Errors on unavailable chain"() { setup: - def upstreams = new UpstreamsMock(Chain.ETHEREUM, Stub(Upstream)) + def upstreams = new UpstreamsMock(Chain.ETHEREUM, Stub(EthereumUpstream)) def streamHead = new StreamHead(upstreams) when: def flux = streamHead.add( diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy similarity index 94% rename from src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy index a96ea73f..0e41d17c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy @@ -33,7 +33,7 @@ import spock.lang.Specification import java.time.Duration -class TrackAddressSpec extends Specification { +class TrackEthereumAddressSpec extends Specification { def chain = Common.ChainRef.CHAIN_ETHEREUM def address1 = "0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f" @@ -59,7 +59,7 @@ class TrackAddressSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackAddress trackAddress = new TrackAddress(upstreams, Schedulers.immediate()) + TrackEthereumAddress trackAddress = new TrackEthereumAddress(upstreams, Schedulers.immediate()) trackAddress.init() apiMock.answer("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") @@ -101,7 +101,7 @@ class TrackAddressSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackAddress trackAddress = new TrackAddress(upstreams, Schedulers.immediate()) + TrackEthereumAddress trackAddress = new TrackEthereumAddress(upstreams, Schedulers.immediate()) trackAddress.init() apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy similarity index 95% rename from src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index 1d9816c8..6d843110 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy @@ -36,7 +36,7 @@ import spock.lang.Specification import java.time.Duration import java.time.Instant -class TrackTxSpec extends Specification { +class TrackEthereumTxSpec extends Specification { def chain = Common.ChainRef.CHAIN_ETHEREUM def txId = "0xba61ce4672751fd6086a9ac2b55547a5555af17535b6c0334ede2ecb6d64070a" @@ -45,8 +45,8 @@ class TrackTxSpec extends Specification { def "Gives details for an old transaction"() { setup: def req = BlockchainOuterClass.TxStatusRequest.newBuilder() - .setChain(chain) - .setConfirmationLimit(6) + .setChain(chain) + .setConfirmationLimit(6) .setTxId(txId) .build() @@ -91,7 +91,7 @@ class TrackTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate()) trackTx.init() apiMock.answer("eth_getTransactionByHash", [txId], txJson) @@ -123,7 +123,7 @@ class TrackTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate()) trackTx.init() apiMock.answer("eth_getTransactionByHash", [txId], null) @@ -180,7 +180,7 @@ class TrackTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate()) trackTx.init() apiMock.answerOnce("eth_getTransactionByHash", [txId], null) @@ -279,7 +279,7 @@ class TrackTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate()) trackTx.init() apiMock.answerOnce("eth_getTransactionByHash", [txId], null) @@ -320,7 +320,7 @@ class TrackTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate()) trackTx.init() def req = BlockchainOuterClass.TxStatusRequest.newBuilder() @@ -341,7 +341,7 @@ class TrackTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - TrackTx trackTx = new TrackTx(upstreams, Schedulers.immediate()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate()) trackTx.init() def req = BlockchainOuterClass.TxStatusRequest.newBuilder() diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 42828cf7..586a7d5b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.test import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods +import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream @@ -33,11 +33,11 @@ class EthereumUpstreamMock extends EthereumUpstream { EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api) { - this(chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain)) + this(chain, api, new DefaultEthereumMethods(TestingCommons.objectMapper(), chain)) } EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull DirectEthereumApi api) { - this(id, chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain)) + this(id, chain, api, new DefaultEthereumMethods(TestingCommons.objectMapper(), chain)) } EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api, CallMethods methods) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 01609999..23043b5b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.ChainUpstreams import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.JacksonRpcConverter @@ -75,7 +76,7 @@ class TestingCommons { } static AggregatedUpstream aggregatedUpstream(EthereumUpstream up) { - return new ChainUpstreams(Chain.ETHEREUM, [up], Caches.default(), objectMapper()) + return new EthereumChainUpstreams(Chain.ETHEREUM, [up], Caches.default(), objectMapper()) } static CachesFactory emptyCaches() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy index 137fada7..f0070cc8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy @@ -18,16 +18,17 @@ package io.emeraldpay.dshackle.test import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.ChainUpstreams -import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods +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.EthereumChainUpstreams import io.emeraldpay.grpc.Chain import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux class UpstreamsMock implements Upstreams { - private Map target = [:] + private Map target = [:] private Map upstreams = [:] UpstreamsMock(Chain chain, Upstream up) { @@ -40,7 +41,7 @@ class UpstreamsMock implements Upstreams { AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { if (!upstreams.containsKey(chain)) { - upstreams[chain] = new ChainUpstreams(chain, [up], Caches.default(), TestingCommons.objectMapper()) + upstreams[chain] = new EthereumChainUpstreams(chain, [up], Caches.default(), TestingCommons.objectMapper()) } else { upstreams[chain].addUpstream(up) } @@ -63,9 +64,9 @@ class UpstreamsMock implements Upstreams { } @Override - QuorumBasedMethods getDefaultMethods(@NotNull Chain chain) { + DefaultEthereumMethods getDefaultMethods(@NotNull Chain chain) { if (target[chain] == null) { - QuorumBasedMethods targets = new QuorumBasedMethods(TestingCommons.objectMapper(), chain) + DefaultEthereumMethods targets = new DefaultEthereumMethods(TestingCommons.objectMapper(), chain) target[chain] = targets } return target[chain] diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy index 15a9e23a..b930c609 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy @@ -21,6 +21,7 @@ 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.DirectEthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams import io.emeraldpay.grpc.Chain import spock.lang.Specification @@ -30,7 +31,7 @@ class AggregatedUpstreamSpec extends Specification { setup: def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new ChainUpstreams(Chain.ETHEREUM, [up1, up2], Caches.default(), TestingCommons.objectMapper()) + def aggr = new EthereumChainUpstreams(Chain.ETHEREUM, [up1, up2], Caches.default(), TestingCommons.objectMapper()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 24bb43fb..0ff1f6e7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -19,7 +19,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.test.EthereumApiStub import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods +import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs @@ -35,7 +35,7 @@ class FilteredApisSpec extends Specification { def rpcClient = Stub(ReactorRpcClient) def objectMapper = TestingCommons.objectMapper() - def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM) + def ethereumTargets = new DefaultEthereumMethods(objectMapper, Chain.ETHEREUM) def "Verifies labels"() { setup: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserverSpec.groovy similarity index 89% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserverSpec.groovy index ca009e16..746833ca 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserverSpec.groovy @@ -13,9 +13,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.emeraldpay.dshackle.upstream +package io.emeraldpay.dshackle.upstream.ethereum -import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead +import io.emeraldpay.dshackle.upstream.HeadLagObserver +import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.rpc.json.BlockJson import reactor.core.publisher.Flux import reactor.core.publisher.TopicProcessor @@ -25,7 +26,7 @@ import spock.lang.Specification import java.time.Duration -class HeadLagObserverSpec extends Specification { +class EthereumHeadLagObserverSpec extends Specification { def "Updates lag distance"() { setup: @@ -64,7 +65,7 @@ class HeadLagObserverSpec extends Specification { 1 * up2.setLag(1) 1 * up2.setLag(0) - HeadLagObserver observer = new HeadLagObserver(master, [up1, up2]) + HeadLagObserver observer = new EthereumHeadLagObserver(master, [up1, up2]) when: def act = observer.subscription().take(Duration.ofMillis(1200)) @@ -78,7 +79,7 @@ class HeadLagObserverSpec extends Specification { def "Probes until there is no difference"() { setup: EthereumHead master = Mock() - HeadLagObserver observer = new HeadLagObserver(master, []) + HeadLagObserver observer = new EthereumHeadLagObserver(master, []) Upstream up = Mock() def blocks = [100, 101, 102].collect { i -> @@ -103,7 +104,7 @@ class HeadLagObserverSpec extends Specification { def "Correct distance"() { setup: EthereumHead master = Mock() - HeadLagObserver observer = new HeadLagObserver(master, []) + HeadLagObserver observer = new EthereumHeadLagObserver(master, []) expect: def top = new BlockJson().with { it.number = topHeight diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy similarity index 96% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index b2a38ef6..58cb81cd 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -34,7 +34,7 @@ import spock.lang.Specification import java.time.Duration import java.util.concurrent.CompletableFuture -class GrpcUpstreamSpec extends Specification { +class EthereumGrpcUpstreamSpec extends Specification { MockServer mockServer = new MockServer() ObjectMapper objectMapper = TestingCommons.objectMapper() @@ -70,7 +70,7 @@ class GrpcUpstreamSpec extends Specification { } }) def transport = ReactorEmeraldClient.newBuilder().connectUsing(client.channel).build() - def upstream = new GrpcUpstream("test", chain, client, objectMapper, transport) + def upstream = new EthereumGrpcUpstream("test", chain, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"]) @@ -126,7 +126,7 @@ class GrpcUpstreamSpec extends Specification { } }) def transport = ReactorEmeraldClient.newBuilder().connectUsing(client.channel).build() - def upstream = new GrpcUpstream("test", Chain.ETHEREUM, client, objectMapper, transport) + def upstream = new EthereumGrpcUpstream("test", Chain.ETHEREUM, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"]) @@ -186,7 +186,7 @@ class GrpcUpstreamSpec extends Specification { } }) def transport = ReactorEmeraldClient.newBuilder().connectUsing(client.channel).build() - def upstream = new GrpcUpstream("test", chain, client, objectMapper, transport) + def upstream = new EthereumGrpcUpstream("test", chain, client, objectMapper, transport) upstream.setLag(0) upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() .addAllSupportedMethods(["eth_getBlockByHash"])