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 ed9e2fb5..5bcce942 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 @@ -99,7 +99,7 @@ open class EthereumPosMultiStream( }?.map { it as GrpcUpstream }?.map { - it.getBlockchainApi().nativeSubscribe(request) + it.proxySubscribe(request) }?.let { Flux.merge(it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 0117efe0..a539a21a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -38,6 +38,7 @@ import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger import java.time.Instant @@ -105,6 +106,9 @@ class BitcoinGrpcUpstream( return remote } + override fun proxySubscribe(request: BlockchainOuterClass.NativeSubscribeRequest): Flux = + remote.nativeSubscribe(request) + override fun getHead(): Head { return grpcHead } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 1459550a..ab4084c5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -41,6 +41,7 @@ import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger import java.time.Instant @@ -110,6 +111,9 @@ open class EthereumGrpcUpstream( return remote } + override fun proxySubscribe(request: BlockchainOuterClass.NativeSubscribeRequest): Flux = + remote.nativeSubscribe(request) + override fun start() { } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 31f4c288..43acab2f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -41,6 +41,7 @@ import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.math.BigInteger import java.time.Instant @@ -130,6 +131,9 @@ open class EthereumPosGrpcUpstream( return remote } + override fun proxySubscribe(request: BlockchainOuterClass.NativeSubscribeRequest): Flux = + remote.nativeSubscribe(request) + // ------------------------------------------------------------------------------------------ override fun getLabels(): Collection { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy index 8b680bd2..5793c17c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy @@ -19,7 +19,6 @@ 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.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe import io.emeraldpay.dshackle.upstream.signature.NoSigner @@ -35,18 +34,21 @@ class NativeSubscribeSpec extends Specification { def "Call with empty params when not provided"() { setup: - def subscribe = Mock(EthereumSubscribe) { - 1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}") - } - def up = Mock(EthereumPosMultiStream) { - 1 * it.getSubscribe() >> subscribe - } - - def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer) def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() .setChainValue(Chain.ETHEREUM.id) .setMethod("newHeads") .build() + + def subscribe = Mock(EthereumSubscribe) { + 1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}") + } + def up = Mock(EthereumPosMultiStream) { + 1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null + 1 * it.getSubscribe() >> subscribe + } + + def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer) + when: def act = nativeSubscribe.start(call) @@ -59,6 +61,15 @@ class NativeSubscribeSpec extends Specification { def "Call with params when provided"() { setup: + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("logs") + .setPayload(ByteString.copyFromUtf8( + '{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' + + '"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}' + )) + .build() + def subscribe = Mock(EthereumSubscribe) { 1 * it.subscribe("logs", { params -> println("params: $params") @@ -71,18 +82,35 @@ class NativeSubscribeSpec extends Specification { }, _ as Selector.AnyLabelMatcher) >> Flux.just("{}") } def up = Mock(EthereumPosMultiStream) { + 1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null 1 * it.getSubscribe() >> subscribe } def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer) + + when: + def act = nativeSubscribe.start(call) + + then: + StepVerifier.create(act) + .expectNext(new NativeSubscribe.ResponseHolder("{}", null)) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Proxy call"() { + setup: def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() .setChainValue(Chain.ETHEREUM.id) - .setMethod("logs") - .setPayload(ByteString.copyFromUtf8( - '{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' + - '"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}' - )) + .setMethod("newHeads") .build() + def up = Mock(EthereumPosMultiStream) { + 1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> Flux.just("{}") + 0 * it.getSubscribe() + } + + def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer) + when: def act = nativeSubscribe.start(call) @@ -93,3 +121,4 @@ class NativeSubscribeSpec extends Specification { .verify(Duration.ofSeconds(1)) } } + diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy index 5d75a195..7c1f4954 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy @@ -52,6 +52,10 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream { this(chain, api, allMethods()) } + EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, Map labels) { + this(id, chain, api, allMethods(), labels) + } + EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api) { this(id, chain, api, allMethods()) } @@ -61,11 +65,15 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream { } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods) { + this(id, chain, api, methods, Collections.emptyMap()) + } + + EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods, Map labels) { super(id, chain, UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.PRIMARY, methods, - new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), + new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)), new ConnectorFactoryMock(api, new EthereumHeadMock())) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 3d3ae1fc..fbec4d77 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.test - import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesFactory @@ -27,15 +26,12 @@ import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.grpc.Chain import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.rpc.json.BlockJson +import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.MeterRegistry import io.micrometer.core.instrument.logging.LoggingMeterRegistry import org.apache.commons.lang3.StringUtils @@ -56,6 +52,10 @@ class TestingCommons { return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api()) } + static EthereumPosRpcUpstreamMock upstream(String id, String provider) { + return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api(), Collections.singletonMap("provider", provider)) + } + static EthereumPosRpcUpstreamMock upstream(String id, Reader api) { return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 42fa96c8..ee9790e9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -16,25 +16,33 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock -import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream +import io.emeraldpay.dshackle.upstream.grpc.EthereumPosGrpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.rpc.json.BlockJson +import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import io.emeraldpay.grpc.Chain import org.jetbrains.annotations.NotNull +import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration import java.time.Instant +import java.time.temporal.ChronoUnit class MultistreamSpec extends Specification { @@ -193,6 +201,162 @@ class MultistreamSpec extends Specification { 1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes) } + def "Filter upstream matching selector single"() { + setup: + + def block1 = createBlock(111111) + def block2 = createBlock(222222) + def block3 = createBlock(222222) + + def up1 = TestingCommons.upstream("test-1", "internal").tap { + it.ethereumHeadMock.predefined = Flux.just(block1) + } + def up2 = TestingCommons.upstream("test-2", "external").tap { + it.ethereumHeadMock.predefined = Flux.just(block2) + } + def up3 = TestingCommons.upstream("test-3", "external").tap { + it.ethereumHeadMock.predefined = Flux.just(block3) + } + def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default()) + + + up1.ethereumHeadMock.toString() + + when: + def head = multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).flux + + then: + StepVerifier.create(head) + .expectNext(block1) + .expectNext(block1) + .expectComplete() + .verify(Duration.ofSeconds(1)) + + } + + def "Proxy gRPC request - select one"() { + setup: + + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("newHeads") + .build() + + def up1 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> true + 1 * getId() >> "internal" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + 1 * proxySubscribe(call) >> Flux.just("{}") + } + def up2 = Mock(EthereumPosGrpcUpstream) { + 1 * getId() >> "external" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "external"))] + } + def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default()) + + when: + def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call) + + then: + StepVerifier.create(act) + .expectNext("{}") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Proxy gRPC request - not all gRPC"() { + setup: + + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("newHeads") + .build() + + def up1 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> true + 1 * getId() >> "1" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + } + def up2 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> false + 1 * getId() >> "2" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + } + def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default()) + + when: + def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call) + + then: + !act + } + + def "Proxy gRPC request - select many"() { + setup: + + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("newHeads") + .build() + + def up1 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> true + 1 * getId() >> "1" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + 1 * proxySubscribe(call) >> Flux.just("{1}") + } + def up2 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> true + 1 * getId() >> "2" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + 1 * proxySubscribe(call) >> Flux.just("{2}") + } + def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default()) + + when: + def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call) + + then: + StepVerifier.create(act) + .expectNextMatches { it as String in ["{1}", "{2}"] } + .expectNextMatches { it as String in ["{1}", "{2}"] } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Proxy gRPC request - select many"() { + setup: + + def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() + .setChainValue(Chain.ETHEREUM.id) + .setMethod("newHeads") + .build() + + def up1 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> true + 1 * getId() >> "1" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + 1 * proxySubscribe(call) >> Flux.just("{1}") + } + def up2 = Mock(EthereumPosGrpcUpstream) { + 1 * isGrpc() >> true + 1 * getId() >> "2" + 1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))] + 1 * proxySubscribe(call) >> Flux.just("{2}") + } + def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default()) + + when: + def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call) + + then: + StepVerifier.create(act) + .expectNextMatches { it as String in ["{1}", "{2}"] } + .expectNextMatches { it as String in ["{1}", "{2}"] } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + class TestMultistream extends Multistream { TestMultistream(List upstreams, @NotNull RequestPostprocessor postprocessor) { @@ -233,4 +397,57 @@ class MultistreamSpec extends Specification { return null } } + + class TestEthereumPosMultistream extends EthereumPosMultiStream { + + TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { + super(chain, upstreams, caches) + } + + @Override + Mono> getRoutedApi(@NotNull Selector.Matcher matcher) { + return null + } + + @Override + Head updateHead() { + return null + } + + @Override + void setHead(@NotNull Head head) { + + } + + @Override + Head getHead() { + return null + } + + public T cast(Class selfType) { + return this + } + + @Override + ChainFees getFeeEstimation() { + return null + } + + @Override + void init() { + + } + } + + BlockContainer createBlock(long number) { + def block = new BlockJson() + block.number = number + block.hash = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000" + number) + block.totalDifficulty = BigInteger.ONE + block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + block.uncles = [] + block.transactions = [] + + return BlockContainer.from(block) + } }