From f508c38dc70a4ff0dcc5985c5d1fc26889e9a252 Mon Sep 17 00:00:00 2001 From: Maxksim Fomenkov Date: Mon, 12 Sep 2022 15:22:48 +0300 Subject: [PATCH] fix tests and lint --- .../dshackle/proxy/WebsocketHandler.kt | 3 ++- .../dshackle/rpc/NativeSubscribe.kt | 2 +- .../dshackle/rpc/TrackERC20Address.kt | 4 +++- .../upstream/ethereum/EthereumMultistream.kt | 3 ++- .../ethereum/subscribe/ConnectBlockUpdates.kt | 3 +-- .../ethereum/subscribe/ConnectNewHeads.kt | 1 - .../ethereum_pos/EthereumPosMultiStream.kt | 23 ++++++++++++++++++- .../proxy/WebsocketHandlerSpec.groovy | 3 ++- .../dshackle/rpc/NativeSubscribeSpec.groovy | 5 ++-- .../dshackle/rpc/TrackERC20AddressSpec.groovy | 4 +++- .../test/MultistreamHolderMock.groovy | 9 ++++++++ .../subscribe/ConnectBlockUpdatesSpec.groovy | 3 ++- .../subscribe/ConnectNewHeadsSpec.groovy | 7 +++--- 13 files changed, 54 insertions(+), 16 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt index 0de59c53..e15be1ba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.proxy import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.BlockchainOuterClass.Selector import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp @@ -139,7 +140,7 @@ class WebsocketHandler( } // produce actual responses val responses = nativeSubscribe - .subscribe(blockchain, methodParams.first, methodParams.second) + .subscribe(blockchain, methodParams.first, methodParams.second, io.emeraldpay.dshackle.upstream.Selector.empty) .map { event -> WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index 73d21734..5bcf4777 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -79,7 +79,7 @@ open class NativeSubscribe( } } - open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher = Selector.empty): Flux { + open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux { val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) return (up as EthereumLikeMultistream) .getSubscribe() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt index 5a968c44..c794fbd6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt @@ -20,6 +20,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.config.TokensConfig import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.etherjar.domain.Address @@ -89,7 +90,8 @@ class TrackERC20Address( .getSubscribe().logs .start( listOf(tokenDefinition.token.contract), - listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")) + listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")), + Selector.empty ) return ethereumAddresses.extract(request.address) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 37fb2583..409d7138 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -163,7 +163,8 @@ open class EthereumMultistream( } } } - }//TODO track unused heads and remove + } + //TODO track unused heads and remove override fun getFeeEstimation(): ChainFees { return feeEstimation diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt index 757650e4..7d635b62 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt @@ -27,10 +27,8 @@ import reactor.core.scheduler.Schedulers import java.time.Duration import java.util.LinkedList import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantReadWriteLock import kotlin.concurrent.read -import kotlin.concurrent.withLock import kotlin.concurrent.write class ConnectBlockUpdates( @@ -50,6 +48,7 @@ class ConnectBlockUpdates( private val historyUpdateLock = ReentrantReadWriteLock() private val connected: MutableMap> = ConcurrentHashMap() + fun connect() = connect(Selector.empty) fun connect(matcher: Selector.Matcher): Flux { return connected.computeIfAbsent(matcher.describeInternal()) { key -> extract(upstream.getHead(matcher)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt index a2886fc8..b91c6e8b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt @@ -49,5 +49,4 @@ class ConnectNewHeads( connected.remove(key) } } - } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index f73d9c9d..4a938c61 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -20,11 +20,13 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.ChainFees +import io.emeraldpay.dshackle.upstream.EmptyHead import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.MergedHead import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -32,6 +34,7 @@ import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.publisher.Mono +import java.util.concurrent.ConcurrentHashMap @Suppress("UNCHECKED_CAST") open class EthereumPosMultiStream( @@ -49,6 +52,7 @@ open class EthereumPosMultiStream( private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory()) private val feeEstimation = EthereumPriorityFees(this, reader, 256) private val subscribe = EthereumSubscribe(this) + private val filteredHeads: MutableMap = ConcurrentHashMap() init { this.init() @@ -69,6 +73,7 @@ open class EthereumPosMultiStream( override fun stop() { super.stop() reader.stop() + filteredHeads.clear() } override fun isRunning(): Boolean { @@ -138,7 +143,23 @@ open class EthereumPosMultiStream( } override fun getHead(mather: Selector.Matcher): Head = - getHead() //TODO + filteredHeads.computeIfAbsent(mather.describeInternal()) { _ -> + upstreams.filter { mather.matches(it) } + .apply { + log.debug("Found $size upstreams matching [${mather.describeInternal()}]") + } + .map { it.getHead() } + .let { + when (it.size) { + 0 -> EmptyHead() + 1 -> it.first() + else -> MergedHead(it, MostWorkForkChoice()).apply { + start() + } + } + } + } + // TODO track unused heads and remove override fun getFeeEstimation(): ChainFees { return feeEstimation diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy index 815bc4a6..235f4778 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.proxy import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.rpc.NativeSubscribe +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.etherjar.rpc.json.RequestJson import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.Counter @@ -108,7 +109,7 @@ class WebsocketHandlerSpec extends Specification { def response2 = [foo: 2] def nativeSubscribe = Mock(NativeSubscribe) { - 1 * it.subscribe(Chain.ETHEREUM, "foo_test", null) >> Flux.fromIterable([response1, response2]) + 1 * it.subscribe(Chain.ETHEREUM, "foo_test", null, Selector.empty) >> Flux.fromIterable([response1, response2]) } def handler = new WebsocketHandler( new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy index 4e176e2a..88743527 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.test.MultistreamHolderMock +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe import io.emeraldpay.grpc.Chain @@ -32,7 +33,7 @@ class NativeSubscribeSpec extends Specification { def "Call with empty params when not provided"() { setup: def subscribe = Mock(EthereumSubscribe) { - 1 * it.subscribe("newHeads", null) >> Flux.just("{}") + 1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}") } def up = Mock(EthereumMultistream) { 1 * it.getSubscribe() >> subscribe @@ -64,7 +65,7 @@ class NativeSubscribeSpec extends Specification { params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65" println("ok: $ok") ok - }) >> Flux.just("{}") + }, _ as Selector.AnyLabelMatcher) >> Flux.just("{}") } def up = Mock(EthereumMultistream) { 1 * it.getSubscribe() >> subscribe diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy index de15a93e..ca189b32 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy @@ -4,6 +4,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.config.TokensConfig import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe @@ -169,7 +170,8 @@ class TrackERC20AddressSpec extends Specification { def logs = Mock(ConnectLogs) { 1 * start( [Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")], - [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")] + [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")], + Selector.empty ) >> { args -> println("ConnectLogs.start $args") Flux.fromIterable(events) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 26275da6..3cecde10 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -20,6 +20,7 @@ package io.emeraldpay.dshackle.test import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -140,6 +141,14 @@ class MultistreamHolderMock implements MultistreamHolder { } return super.getHead() } + + @Override + Head getHead(@NotNull Selector.Matcher mather) { + if (customHead != null) { + return customHead + } + return super.getHead(mather) + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy index 697fd263..e9557335 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId @@ -223,7 +224,7 @@ class ConnectBlockUpdatesSpec extends Specification { 1 * getFlux() >> Flux.never() } def up = Mock(EthereumMultistream) { - 1 * getHead() >> head + 1 * getHead(Selector.empty) >> head } def connectBlockUpdates = new ConnectBlockUpdates(up) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy index 6a62bf6b..25781ccc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import reactor.core.publisher.Flux import reactor.test.StepVerifier @@ -17,12 +18,12 @@ class ConnectNewHeadsSpec extends Specification { ]) } def up = Mock(EthereumMultistream) { - 1 * getHead() >> head + 1 * getHead(Selector.empty) >> head } ConnectNewHeads connectNewHeads = new ConnectNewHeads(up) when: - def act1 = connectNewHeads.connect() - def act2 = connectNewHeads.connect() + def act1 = connectNewHeads.connect(Selector.empty) + def act2 = connectNewHeads.connect(Selector.empty) then: StepVerifier.create(act1) .expectNextCount(1)