problem: routed api is not executed first

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -19,14 +19,17 @@ 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.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.RpcResponseError
 import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -116,17 +119,25 @@ open class NativeCall(
     }
 
     fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
-//        ctx.upstream.getRoutedApi(ctx.matcher)
+        return ctx.upstream.getRoutedApi(ctx.matcher)
-//                .flatMap { api ->
+                .flatMap { api ->
-//                    api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
+                    api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
-//                }.switchIfEmpty(
+                            .flatMap(JsonRpcResponse::requireResult)
-//                        Mono.just(ctx).flatMap(this::executeOnRemote)
+                            .map {
-//                )
+                                ctx.withPayload(it)
-        //TODO use routed api
+                            }
-        return executeOnRemote(ctx)
+                }.switchIfEmpty(
+                        Mono.just(ctx).flatMap(this::executeOnRemote)
+                )
+                .onErrorMap {
+                    CallFailure(ctx.id, it)
+                }
     }
 
     fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
+        if (!ctx.upstream.getMethods().isAllowed(ctx.payload.method)) {
+            return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
+        }
         //TODO move to routed api
         val apis = ctx.getApis()
         apis.request(1)

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -101,7 +101,7 @@ abstract class Multistream(
         val apis = getApiSource(matcher)
         apis.request(1)
         return Mono.from(apis)
-                .switchIfEmpty(Mono.error(Exception("No API available")))
+                .switchIfEmpty(Mono.error(Exception("No API available for $chain")))
     }
 
     /**

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.reader.EmptyReader
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -74,7 +75,7 @@ open class BitcoinMultistream(
 
     override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
         //TODO
-        return getDirectApi(matcher)
+        return Mono.just(EmptyReader())
     }
 
     open fun getReader(): BitcoinReader {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt

@@ -119,9 +119,7 @@ open class EthereumMultistream(
     }
 
     override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
-        return getDirectApi(matcher).map { api ->
+        return Mono.just(NativeCallRouter(objectMapper, reader, getMethods()))
-            NativeCallRouter(objectMapper, reader, api, getMethods())
-        }
     }
 
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt

@@ -32,7 +32,6 @@ import java.math.BigInteger
 class NativeCallRouter(
         private val objectMapper: ObjectMapper,
         private val reader: EthereumReader,
-        private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
         private val methods: CallMethods
 ) : Reader<JsonRpcRequest, JsonRpcResponse> {
 

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy

@@ -19,11 +19,16 @@ package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.quorum.BroadcastQuorum
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
+import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.ethereum.NativeCallRouter
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.rpc.ReactorRpcClient
 import io.infinitape.etherjar.rpc.RpcException
@@ -40,6 +45,55 @@ class NativeCallSpec extends Specification {
 
     def objectMapper = TestingCommons.objectMapper()
 
+    def "Tries router first"() {
+        def routedApi = Mock(Reader) {
+            1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(new JsonRpcResponse("1".bytes, null))
+        }
+        def upstream = Mock(Multistream) {
+            1 * getRoutedApi(_) >> Mono.just(routedApi)
+        }
+        def upstreams = Stub(MultistreamHolder)
+
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(
+                1, upstream, Selector.empty, new AlwaysQuorum(),
+                new NativeCall.ParsedCallDetails("eth_test", [])
+        )
+
+        when:
+        def act = nativeCall.fetch(ctx).block(Duration.ofSeconds(1))
+        then:
+        act.payload == "1".bytes
+    }
+
+    def "Return error if router denied the requests"() {
+        def routedApi = Mock(Reader) {
+            1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.error(new RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Test message"))
+        }
+        def upstream = Mock(Multistream) {
+            1 * getRoutedApi(_) >> Mono.just(routedApi)
+        }
+        def upstreams = Stub(MultistreamHolder)
+
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(
+                15, upstream, Selector.empty, new AlwaysQuorum(),
+                new NativeCall.ParsedCallDetails("eth_test", [])
+        )
+
+        when:
+        def act = nativeCall.fetch(ctx) //.block(Duration.ofSeconds(1))
+        then:
+        StepVerifier.create(act)
+                .expectErrorMatches { t ->
+                    t instanceof NativeCall.CallFailure &&
+                            t.id == 15 &&
+                            t.reason instanceof RpcException &&
+                            t.reason.rpcMessage == "Test message"
+                }
+                .verify(Duration.ofSeconds(1))
+    }
+
     def "Quorum is applied"() {
         setup:
         def quorum = Spy(new AlwaysQuorum())

src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy

@@ -16,13 +16,16 @@
  */
 package io.emeraldpay.dshackle.test
 
-import io.emeraldpay.dshackle.cache.Caches
+
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
+import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -36,12 +39,20 @@ class EthereumUpstreamMock extends EthereumUpstream {
 
     EthereumHeadMock ethereumHeadMock = new EthereumHeadMock()
 
+    static CallMethods allMethods() {
+        new AggregatedCallMethods([
+                new DefaultEthereumMethods(TestingCommons.objectMapper(), Chain.ETHEREUM),
+                new DefaultBitcoinMethods(TestingCommons.objectMapper()),
+                new DirectCallMethods(["eth_test"])
+        ])
+    }
+
     EthereumUpstreamMock(@NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api) {
-        this(chain, api, new DefaultEthereumMethods(TestingCommons.objectMapper(), chain))
+        this(chain, api, allMethods())
     }
 
     EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api) {
-        this(id, chain, api, new DefaultEthereumMethods(TestingCommons.objectMapper(), chain))
+        this(id, chain, api, allMethods())
     }
 
     EthereumUpstreamMock(@NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy

@@ -23,7 +23,6 @@ class NativeCallRouterSpec extends Specification {
                         Caches.default(TestingCommons.objectMapper()),
                         TestingCommons.objectMapper()
                 ),
-                new EmptyReader<JsonRpcRequest, JsonRpcResponse>(),
                 methods
         )
         when: