problem: quorum is not applied for internal reads

src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt

@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.upstream.ApiSource
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 
 // creates instance of a Quorum based reader
-interface QuorumReaderFactory {
+open interface QuorumReaderFactory {
 
     companion object {
         fun default(): QuorumReaderFactory {
@@ -28,7 +28,7 @@ interface QuorumReaderFactory {
         }
     }
 
-    fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result>
+    open fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result>
 
     class Default : QuorumReaderFactory {
         override fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result> {

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

@@ -16,9 +16,6 @@
  */
 package io.emeraldpay.dshackle.upstream
 
-import io.emeraldpay.dshackle.reader.Reader
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.reactivestreams.Publisher
 
 interface ApiSource : Publisher<Upstream> {

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

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.cache.CachesFactory
+import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
 import io.emeraldpay.dshackle.startup.UpstreamChange
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream

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

@@ -24,6 +24,8 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
+import org.apache.commons.collections4.Factory
+import org.apache.commons.collections4.FunctorException
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
@@ -52,6 +54,9 @@ abstract class Multistream(
     private var cacheSubscription: Disposable? = null
     private val reconfigLock = ReentrantLock()
     private var callMethods: CallMethods? = null
+    private var callMethodsFactory: Factory<CallMethods> = Factory {
+        return@Factory callMethods ?: throw FunctorException("Not initialized yet")
+    }
     private var seq = 0
     protected var lagObserver: HeadLagObserver? = null
     private var subscription: Disposable? = null
@@ -86,7 +91,7 @@ abstract class Multistream(
     /**
      * Get a source for direct APIs
      */
-    fun getApiSource(matcher: Selector.Matcher): ApiSource {
+    open fun getApiSource(matcher: Selector.Matcher): ApiSource {
         val i = seq++
         if (seq >= Int.MAX_VALUE / 2) {
             seq = 0
@@ -117,6 +122,7 @@ abstract class Multistream(
     fun onUpstreamsUpdated() {
         reconfigLock.withLock {
             getAll().map { it.getMethods() }.let {
+                //TODO made list of uniq instances, and then if only one, just use it directly
                 callMethods = AggregatedCallMethods(it)
             }
         }
@@ -147,6 +153,10 @@ abstract class Multistream(
         return callMethods ?: throw IllegalStateException("Methods are not initialized yet")
     }
 
+    fun getMethodsFactory(): Factory<CallMethods> {
+        return callMethodsFactory
+    }
+
     override fun start() {
         subscription = observeStatus()
                 .distinctUntilChanged()

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

@@ -0,0 +1,134 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CurrentBlockCache
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.TxContainer
+import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.infinitape.etherjar.domain.Address
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.domain.Wei
+import io.infinitape.etherjar.hex.HexQuantity
+import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.RpcResponseError
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import org.apache.commons.collections4.Factory
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import reactor.util.retry.Retry
+import java.time.Duration
+import java.util.concurrent.TimeoutException
+
+/**
+ * Common reads from upstream, makes actual calls with applying quorum and retries
+ */
+class EthereumDirectReader(
+        private val up: Multistream,
+        private val caches: Caches,
+        private val balanceCache: CurrentBlockCache<Address, Wei>,
+        private val callMethodsFactory: Factory<CallMethods>
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EthereumDirectReader::class.java)
+    }
+
+    private val objectMapper: ObjectMapper = Global.objectMapper
+    open var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()
+
+    val blockReader: Reader<BlockHash, BlockContainer>
+    val blockByHeightReader: Reader<Long, BlockContainer>
+    val txReader: Reader<TransactionId, TxContainer>
+    val balanceReader: Reader<Address, Wei>
+
+    init {
+        blockReader = object : Reader<BlockHash, BlockContainer> {
+            override fun read(key: BlockHash): Mono<BlockContainer> {
+                val request = JsonRpcRequest("eth_getBlockByHash", listOf(key.toHex(), false))
+                return readBlock(request, key.toHex())
+            }
+        }
+        blockByHeightReader = object : Reader<Long, BlockContainer> {
+            override fun read(key: Long): Mono<BlockContainer> {
+                val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false))
+                return readBlock(request, key.toString())
+            }
+        }
+        txReader = object : Reader<TransactionId, TxContainer> {
+            override fun read(key: TransactionId): Mono<TxContainer> {
+                val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
+                return readWithQuorum(request)
+                        .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
+                        .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                        .flatMap { txbytes ->
+                            val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
+                            if (tx == null) {
+                                Mono.empty()
+                            } else {
+                                Mono.just(TxContainer.from(tx, txbytes))
+                            }
+                        }
+                        .doOnNext { tx ->
+                            if (tx.blockId != null) {
+                                caches.cache(Caches.Tag.REQUESTED, tx)
+                            }
+                        }
+
+            }
+        }
+        balanceReader = object : Reader<Address, Wei> {
+            override fun read(key: Address): Mono<Wei> {
+                val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), "latest"))
+                return readWithQuorum(request)
+                        .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
+                        .map {
+                            val str = String(it)
+                            // it's a json string, i.e. wrapped with quotes, ex. _"0x1234"_
+                            if (str.startsWith("\"") && str.endsWith("\"")) {
+                                Wei.from(str.substring(1, str.length - 1))
+                            } else {
+                                throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value")
+                            }
+                        }
+                        .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                        .doOnNext { value ->
+                            balanceCache.put(key, value)
+                        }
+            }
+        }
+    }
+
+    private fun readBlock(request: JsonRpcRequest, id: String): Mono<BlockContainer> {
+        return readWithQuorum(request)
+                .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id")))
+                .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                .map { blockbytes ->
+                    val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>
+                    BlockContainer.from(block, blockbytes)
+                }
+                .doOnNext { block ->
+                    caches.cache(Caches.Tag.REQUESTED, block)
+                }
+    }
+
+    /**
+     * Read from an Upstream applying a Quorum specific for that request
+     */
+    private fun readWithQuorum(request: JsonRpcRequest): Mono<ByteArray> {
+        return quorumReaderFactory
+                .create(up.getApiSource(Selector.empty), callMethodsFactory.create().getQuorumFor(request.method))
+                .read(request)
+                .map { it.value }
+    }
+}

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

@@ -16,8 +16,6 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import com.fasterxml.jackson.databind.ObjectMapper
-import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
@@ -39,10 +37,9 @@ open class EthereumMultistream(
         private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)
     }
 
-    private val objectMapper: ObjectMapper = Global.objectMapper
     private var head: Head? = null
 
-    private val reader: EthereumReader = EthereumReader(this, this.caches)
+    private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
 
     init {
         this.init()

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

@@ -16,38 +16,32 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import com.fasterxml.jackson.databind.ObjectMapper
-import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CurrentBlockCache
 import io.emeraldpay.dshackle.data.*
 import io.emeraldpay.dshackle.reader.*
 import io.emeraldpay.dshackle.upstream.Multistream
-import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.infinitape.etherjar.domain.Address
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.domain.Wei
-import io.infinitape.etherjar.hex.HexQuantity
-import io.infinitape.etherjar.rpc.RpcException
-import io.infinitape.etherjar.rpc.RpcResponseError
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import org.apache.commons.collections4.Factory
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
-import reactor.core.Disposable
-import reactor.core.publisher.Mono
-import reactor.util.retry.Retry
-import java.time.Duration
-import java.util.concurrent.TimeoutException
 import java.util.function.Function
 
+/**
+ * Reader for the common operations, that wraps caches + native call with quorum verification
+ */
 open class EthereumReader(
         private val up: Multistream,
-        private val caches: Caches
+        private val caches: Caches,
+        private val callMethodsFactory: Factory<CallMethods>
 ) : Lifecycle {
 
     companion object {
@@ -56,6 +50,7 @@ open class EthereumReader(
 
     private val objectMapper: ObjectMapper = Global.objectMapper
     private val balanceCache = CurrentBlockCache<Address, Wei>()
+    private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory)
 
     val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block ->
         val existing = block.getParsed(BlockJson::class.java)
@@ -87,115 +82,17 @@ open class EthereumReader(
         TxContainer.from(tx)
     }
 
-    private val blocksDirect: Reader<BlockHash, BlockContainer>
-    private val blocksByHeightDirect: Reader<Long, BlockContainer>
-    private val txDirect: Reader<TransactionId, TxContainer>
-    private val balanceDirect: Reader<Address, Wei>
-
     private val idToBlockHash = Function<BlockId, BlockHash> { id -> BlockHash.from(id.value) }
     private val blockHashToId = Function<BlockHash, BlockId> { hash -> BlockId.from(hash) }
 
     private val txHashToId = Function<TransactionId, TxId> { hash -> TxId.from(hash) }
     private val idToTxHash = Function<TxId, TransactionId> { id -> TransactionId.from(id.value) }
 
-    private val directResponseBytes = Function<JsonRpcResponse, ByteArray> { resp ->
-        if (resp.error != null) {
-            throw resp.error.asException()
-        } else {
-            resp.getResult()
-        }
-    }
-
-    init {
-        blocksDirect = object : Reader<BlockHash, BlockContainer> {
-            override fun read(key: BlockHash): Mono<BlockContainer> {
-                return up.getDirectApi(Selector.empty).flatMap { api ->
-                    val request = JsonRpcRequest("eth_getBlockByHash", listOf(key.toHex(), false))
-                    api.read(request)
-                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $key")))
-                            .map(directResponseBytes)
-                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
-                            .map { blockbytes ->
-                                val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>
-                                BlockContainer.from(block, blockbytes)
-                            }
-                            .doOnNext { block ->
-                                caches.cache(Caches.Tag.REQUESTED, block)
-                            }
-                }
-            }
-        }
-        blocksByHeightDirect = object : Reader<Long, BlockContainer> {
-            override fun read(key: Long): Mono<BlockContainer> {
-                return up.getDirectApi(Selector.empty).flatMap { api ->
-                    val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false))
-                    api.read(request)
-                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $key")))
-                            .map(directResponseBytes)
-                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
-                            .map { blockbytes ->
-                                val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>
-                                BlockContainer.from(block, blockbytes)
-                            }
-                            .doOnNext { block ->
-                                caches.cache(Caches.Tag.REQUESTED, block)
-                            }
-                }
-            }
-        }
-        txDirect = object : Reader<TransactionId, TxContainer> {
-            override fun read(key: TransactionId): Mono<TxContainer> {
-                return up.getDirectApi(Selector.empty).flatMap { api ->
-                    val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
-                    api.read(request)
-                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
-                            .map(directResponseBytes)
-                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
-                            .flatMap { txbytes ->
-                                val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
-                                if (tx == null) {
-                                    Mono.empty()
-                                } else {
-                                    Mono.just(TxContainer.from(tx, txbytes))
-                                }
-                            }
-                            .doOnNext { tx ->
-                                if (tx.blockId != null) {
-                                    caches.cache(Caches.Tag.REQUESTED, tx)
-                                }
-                            }
-                }
-            }
-        }
-        balanceDirect = object : Reader<Address, Wei> {
-            override fun read(key: Address): Mono<Wei> {
-                return up.getDirectApi(Selector.empty).flatMap { api ->
-                    val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), "latest"))
-                    api.read(request)
-                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
-                            .map(directResponseBytes)
-                            .map {
-                                val str = String(it)
-                                if (str.startsWith("\"") && str.endsWith("\"")) {
-                                    Wei.from(str.substring(1, str.length - 1))
-                                } else {
-                                    throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value")
-                                }
-                            }
-                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
-                            .doOnNext { value ->
-                                balanceCache.put(key, value)
-                            }
-                }
-            }
-        }
-    }
-
     fun blocksByHash(): Reader<BlockHash, BlockJson<TransactionRefJson>> {
         return TransformingReader(
                 CompoundReader(
                         RekeyingReader(blockHashToId, caches.getBlocksByHash()),
-                        blocksDirect
+                        directReader.blockReader
                 ),
                 extractBlock
         )
@@ -205,7 +102,7 @@ open class EthereumReader(
         return TransformingReader(
                 CompoundReader(
                         caches.getBlocksByHash(),
-                        RekeyingReader(idToBlockHash, blocksDirect)
+                        RekeyingReader(idToBlockHash, directReader.blockReader)
                 ),
                 extractBlock
         )
@@ -228,7 +125,7 @@ open class EthereumReader(
     fun blocksByHeightAsCont(): Reader<Long, BlockContainer> {
         return CompoundReader(
                 caches.getBlocksByHeight(),
-                blocksByHeightDirect
+                directReader.blockByHeightReader
         )
     }
 
@@ -236,7 +133,7 @@ open class EthereumReader(
         return TransformingReader(
                 CompoundReader(
                         RekeyingReader(txHashToId, caches.getTxByHash()),
-                        txDirect
+                        directReader.txReader
                 ),
                 extractTx
         )
@@ -245,13 +142,13 @@ open class EthereumReader(
     fun txByHashAsCont(): Reader<TxId, TxContainer> {
         return CompoundReader(
                 caches.getTxByHash(),
-                RekeyingReader(idToTxHash, txDirect)
+                RekeyingReader(idToTxHash, directReader.txReader)
         )
     }
 
     fun balance(): Reader<Address, Wei> {
         return CompoundReader(
-                balanceCache, balanceDirect
+                balanceCache, directReader.balanceReader
         )
     }
 

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

@@ -0,0 +1,174 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CurrentBlockCache
+import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
+import io.emeraldpay.dshackle.quorum.QuorumRpcReader
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.ApiSource
+import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.Address
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.domain.Wei
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionJson
+import org.apache.commons.collections4.Factory
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+import java.time.Instant
+
+class EthereumDirectReaderSpec extends Specification {
+
+    String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
+    String address1 = "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d"
+
+    def "Reads block by hash"() {
+        setup:
+        def json = new BlockJson().tap {
+            number = 100
+            hash = BlockHash.from(hash1)
+            timestamp = Instant.now()
+            totalDifficulty = BigInteger.ONE
+            transactions = []
+        }
+        def up = Mock(Multistream) {
+            1 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        EthereumDirectReader reader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls
+        )
+        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            1 * create(_, _) >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
+                        new QuorumRpcReader.Result(
+                                Global.objectMapper.writeValueAsBytes(json), 1
+                        )
+                )
+            }
+        }
+        when:
+        def act = reader.blockReader.read(BlockHash.from(hash1))
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { block ->
+                    block.hash.toHexWithPrefix() == hash1
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Reads block by height"() {
+        setup:
+        def json = new BlockJson().tap {
+            number = 100
+            hash = BlockHash.from(hash1)
+            timestamp = Instant.now()
+            totalDifficulty = BigInteger.ONE
+            transactions = []
+        }
+        def up = Mock(Multistream) {
+            1 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        EthereumDirectReader reader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls
+        )
+        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            1 * create(_, _) >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just(
+                        new QuorumRpcReader.Result(
+                                Global.objectMapper.writeValueAsBytes(json), 1
+                        )
+                )
+            }
+        }
+        when:
+        def act = reader.blockByHeightReader.read(100)
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { block ->
+                    block.hash.toHexWithPrefix() == hash1
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Reads tx"() {
+        setup:
+        def json = new TransactionJson().tap {
+            hash = TransactionId.from(hash1)
+            blockNumber = 100
+            blockHash = BlockHash.from(hash1)
+        }
+        def up = Mock(Multistream) {
+            1 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        EthereumDirectReader reader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls
+        )
+        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            1 * create(_, _) >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
+                        new QuorumRpcReader.Result(
+                                Global.objectMapper.writeValueAsBytes(json), 1
+                        )
+                )
+            }
+        }
+        when:
+        def act = reader.txReader.read(TransactionId.from(hash1))
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { block ->
+                    block.hash.toHexWithPrefix() == hash1
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Reads balance"() {
+        setup:
+        def up = Mock(Multistream) {
+            1 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        EthereumDirectReader reader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls
+        )
+        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            1 * create(_, _) >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just(
+                        new QuorumRpcReader.Result(
+                                Global.objectMapper.writeValueAsBytes("0x100"), 1
+                        )
+                )
+            }
+        }
+        when:
+        def act = reader.balanceReader.read(Address.from(address1))
+        then:
+        StepVerifier.create(act)
+                .expectNext(Wei.from("0x100"))
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+}

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

@@ -24,6 +24,8 @@ import io.emeraldpay.dshackle.data.TxContainer
 import io.emeraldpay.dshackle.test.EthereumUpstreamMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.Address
 import io.infinitape.etherjar.domain.BlockHash
@@ -32,6 +34,8 @@ import io.infinitape.etherjar.domain.Wei
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import org.apache.commons.collections4.Factory
+import org.apache.commons.collections4.functors.ConstantFactory
 import reactor.core.publisher.Mono
 import spock.lang.Specification
 
@@ -54,6 +58,7 @@ class EthereumReaderSpec extends Specification {
         json.blockHash = blockJson.hash
         json.blockNumber = blockJson.number
     }
+    Factory<CallMethods> calls = ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
 
     def "Block by Id reads from cache"() {
         setup:
@@ -63,7 +68,7 @@ class EthereumReaderSpec extends Specification {
         def caches = Caches.newBuilder()
                 .setBlockByHash(memCache)
                 .build()
-        def reader = new EthereumReader(Stub(Multistream), caches)
+        def reader = new EthereumReader(Stub(Multistream), caches, calls)
 
         when:
         def act = reader.blocksById().read(blockId).block()
@@ -84,7 +89,7 @@ class EthereumReaderSpec extends Specification {
         api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
 
         def upstream = TestingCommons.aggregatedUpstream(api)
-        def reader = new EthereumReader(upstream, caches)
+        def reader = new EthereumReader(upstream, caches, calls)
 
         when:
         def act = reader.blocksById().read(blockId).block()
@@ -105,7 +110,7 @@ class EthereumReaderSpec extends Specification {
         api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
 
         def upstream = TestingCommons.aggregatedUpstream(api)
-        def reader = new EthereumReader(upstream, caches)
+        def reader = new EthereumReader(upstream, caches, calls)
 
         when:
         def act = reader.blocksById().read(blockId).block()
@@ -122,7 +127,7 @@ class EthereumReaderSpec extends Specification {
         def caches = Caches.newBuilder()
                 .setBlockByHash(memCache)
                 .build()
-        def reader = new EthereumReader(Stub(Multistream), caches)
+        def reader = new EthereumReader(Stub(Multistream), caches, calls)
 
         when:
         def act = reader.blocksByHash().read(blockJson.hash).block()
@@ -142,7 +147,7 @@ class EthereumReaderSpec extends Specification {
         def api = TestingCommons.api()
         api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
         def upstream = TestingCommons.aggregatedUpstream(api)
-        def reader = new EthereumReader(upstream, caches)
+        def reader = new EthereumReader(upstream, caches, calls)
 
         when:
         def act = reader.blocksByHash().read(blockJson.hash).block()
@@ -159,7 +164,7 @@ class EthereumReaderSpec extends Specification {
         def caches = Caches.newBuilder()
                 .setTxByHash(memCache)
                 .build()
-        def reader = new EthereumReader(Stub(Multistream), caches)
+        def reader = new EthereumReader(Stub(Multistream), caches, calls)
 
         when:
         def act = reader.txByHash().read(txJson.hash).block()
@@ -180,7 +185,7 @@ class EthereumReaderSpec extends Specification {
         def api = TestingCommons.api()
         api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson)
         def upstream = TestingCommons.aggregatedUpstream(api)
-        def reader = new EthereumReader(upstream, caches)
+        def reader = new EthereumReader(upstream, caches, calls)
 
         when:
         def act = reader.txByHash().read(txJson.hash).block()
@@ -196,7 +201,7 @@ class EthereumReaderSpec extends Specification {
         api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff")
         EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api)
         def upstreams = TestingCommons.aggregatedUpstream(upstream)
-        def reader = new EthereumReader(upstreams, Caches.default())
+        def reader = new EthereumReader(upstreams, Caches.default(), calls)
         reader.start()
 
         when:

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

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.grpc.Chain
+import org.apache.commons.collections4.functors.ConstantFactory
 import spock.lang.Specification
 
 import java.time.Duration
@@ -17,7 +18,8 @@ class NativeCallRouterSpec extends Specification {
         def router = new NativeCallRouter(
                 new EthereumReader(
                         TestingCommons.aggregatedUpstream(TestingCommons.api()),
-                        Caches.default()
+                        Caches.default(),
+                        ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
                 ),
                 methods
         )