tests

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)
         }

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<out Any> =
+        remote.nativeSubscribe(request)
+
     override fun getHead(): Head {
         return grpcHead
     }

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<out Any> =
+        remote.nativeSubscribe(request)
+
     override fun start() {
     }
 

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<out Any> =
+        remote.nativeSubscribe(request)
+
     // ------------------------------------------------------------------------------------------
 
     override fun getLabels(): Collection<UpstreamsConfig.Labels> {

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")
+                .setMethod("newHeads")
-                .setPayload(ByteString.copyFromUtf8(
-                        '{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' +
-                                '"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}'
-                ))
                 .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))
     }
 }
+

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<JsonRpcRequest, JsonRpcResponse> api, Map<String, String> labels) {
+        this(id, chain, api, allMethods(), labels)
+    }
+
     EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api) {
         this(id, chain, api, allMethods())
     }
@@ -61,11 +65,15 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
     }
 
     EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
+        this(id, chain, api, methods, Collections.<String, String>emptyMap())
+    }
+
+    EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> 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)

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<JsonRpcRequest, JsonRpcResponse> api) {
         return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api)
     }

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<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) {
@@ -233,4 +397,57 @@ class MultistreamSpec extends Specification {
             return null
         }
     }
+
+    class TestEthereumPosMultistream extends EthereumPosMultiStream {
+
+        TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
+            super(chain, upstreams, caches)
+        }
+
+        @Override
+        Mono<Reader<JsonRpcRequest, JsonRpcResponse>> 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 extends Upstream> T cast(Class<T> selfType) {
+            return this
+        }
+
+        @Override
+        ChainFees getFeeEstimation() {
+            return null
+        }
+
+        @Override
+        void init() {
+
+        }
+    }
+
+    BlockContainer createBlock(long number) {
+        def block = new BlockJson<TransactionRefJson>()
+        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)
+    }
 }