problem: different places to fetch from upstream

solution: single access reader for Ethereum upstreams
2020-05-04 21:51:42 -04:00
parent 3a13fb3bd5
commit 0ef46b2980
25 changed files with 748 additions and 67 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/CurrentBlockCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CurrentBlockCache.kt
@@ -0,0 +1,24 @@
+package io.emeraldpay.dshackle.cache
+
+import io.emeraldpay.dshackle.reader.Reader
+import reactor.core.publisher.Mono
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicReference
+
+class CurrentBlockCache<K, D> : Reader<K, D> {
+
+    private val cache = AtomicReference(ConcurrentHashMap<K, D>())
+
+    override fun read(key: K): Mono<D> {
+        return Mono.justOrEmpty(cache.get()[key])
+    }
+
+    fun put(key: K, data: D) {
+        cache.get()[key] = data
+    }
+
+    fun evict() {
+        cache.set(ConcurrentHashMap())
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt
@@ -16,8 +16,11 @@
 */
 package io.emeraldpay.dshackle.reader

+import io.emeraldpay.dshackle.Defaults
+import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.time.Duration

 /**
 * Composition of multiple readers. Reader returns first value returned by any of the source readers.
@@ -26,12 +29,21 @@ class CompoundReader<K, D>(
        private vararg val readers: Reader<K, D>
 ): Reader<K, D> {

+    companion object {
+        private val log = LoggerFactory.getLogger(CompoundReader::class.java)
+    }
+
    override fun read(key: K): Mono<D> {
        if (readers.isEmpty()) {
            return Mono.empty()
        }
        return Flux.fromIterable(readers.asIterable())
-                .flatMap { it.read(key) }.next()
+                .flatMap { rdr ->
+                    rdr.read(key)
+                            .timeout(Defaults.timeoutInternal, Mono.empty())
+                            .doOnError { t -> log.warn("Failed to read from $rdr", t) }
+                            .onErrorResume { Mono.empty() }
+                }.next()
    }

 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt
@@ -0,0 +1,43 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.reader
+
+import reactor.core.publisher.Mono
+import java.util.function.Function
+
+/**
+ * Reader wrapper that maps the input key from ne value to another (ex. convert from Long to String)
+ */
+class RekeyingReader<K, K1, D>(
+        /**
+         * Mapping between original Key and Key supported by the reader
+         */
+        private val rekey: Function<K, K1>,
+        /**
+         * Actual reader
+         */
+        private val reader: Reader<K1, D>
+) : Reader<K, D> {
+
+    override fun read(key: K): Mono<D> {
+        return Mono.just(key)
+                .map(rekey)
+                .flatMap {
+                    reader.read(it)
+                }
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt
@@ -0,0 +1,39 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.reader
+
+import reactor.core.publisher.Mono
+import java.util.function.Function
+
+/**
+ * Reader wrapper that transforms output of the reader to a different format
+ */
+class TransformingReader<K, D0, D>(
+        /**
+         * Actual reader
+         */
+        private val reader: Reader<K, D0>,
+        /**
+         * Result transformation
+         */
+        private val transformer: Function<in D0, out D>
+) : Reader<K, D> {
+
+    override fun read(key: K): Mono<D> {
+        return reader.read(key).map(transformer)
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
@@ -51,24 +51,39 @@ class BlockchainRpc(
    override fun subscribeTxStatus(request: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
        return request.flatMapMany { request ->
            val chain = Chain.byId(request.chainValue)
-            trackTx.find { it.isSupported(chain) }?.subscribe(request)
-                    ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+            try {
+                trackTx.find { it.isSupported(chain) }?.subscribe(request)
+                        ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+            } catch (t: Throwable) {
+                log.error("Internal error during Tx Subscription", t)
+                Flux.error<BlockchainOuterClass.TxStatus>(IllegalStateException("Internal Error"))
+            }
        }
    }

    override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
        return requestMono.flatMapMany { request ->
            val chain = Chain.byId(request.asset.chainValue)
-            trackAddress.find { it.isSupported(chain) }?.subscribe(request)
-                    ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+            try {
+                trackAddress.find { it.isSupported(chain) }?.subscribe(request)
+                        ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+            } catch (t: Throwable) {
+                log.error("Internal error during Balance Subscription", t)
+                Flux.error<BlockchainOuterClass.AddressBalance>(IllegalStateException("Internal Error"))
+            }
        }
    }

    override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
        return requestMono.flatMapMany { request ->
            val chain = Chain.byId(request.asset.chainValue)
-            trackAddress.find { it.isSupported(chain) }?.getBalance(request)
-                    ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+            try {
+                trackAddress.find { it.isSupported(chain) }?.getBalance(request)
+                        ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
+            } catch (t: Throwable) {
+                log.error("Internal error during Balance Request", t)
+                Flux.error<BlockchainOuterClass.AddressBalance>(IllegalStateException("Internal Error"))
+            }
        }
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt
@@ -128,7 +128,6 @@ class TrackBitcoinTx(
    }

    fun loadMempool(upstream: BitcoinUpstream, txid: String): Mono<TxStatus> {
-        println("access: ${upstream.getData()}")
        val mempool = upstream.getData().getMempool().get()
        return mempool.map {
            if (it.contains(txid)) {
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt
@@ -21,15 +21,12 @@ import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.SilentException
-import io.emeraldpay.dshackle.upstream.AggregatedUpstream
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
+import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.Address
 import io.infinitape.etherjar.domain.Wei
-import io.infinitape.etherjar.rpc.Commands
-import io.infinitape.etherjar.rpc.json.BlockTag
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
@@ -90,6 +87,11 @@ class TrackEthereumAddress(
        }
    }

+    fun getUpstream(chain: Chain): AggregatedEthereumUpstreams {
+        return upstreams.getUpstream(chain)?.cast(AggregatedEthereumUpstreams::class.java, EthereumApi::class.java)
+                ?: throw SilentException.UnsupportedBlockchain(chain)
+    }
+
    private fun initAddress(request: BlockchainOuterClass.BalanceRequest): Flux<TrackedAddress> {
        val chain = Chain.byId(request.asset.chainValue)
        if (!upstreams.isAvailable(chain)) {
@@ -120,10 +122,10 @@ class TrackEthereumAddress(
    }

    fun getBalance(addr: TrackedAddress): Mono<Wei> {
-        val up = upstreams.getUpstream(addr.chain) as AggregatedUpstream<EthereumApi>?
-                ?: return Mono.error(SilentException.UnsupportedBlockchain(addr.chain))
-        return up.getApi(Selector.empty)
-                .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) }
+        return getUpstream(addr.chain)
+                .getReader()
+                .balance()
+                .read(addr.address)
                .timeout(Defaults.timeout)
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
@@ -23,15 +23,13 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.TxId
-import io.emeraldpay.dshackle.upstream.AggregatedUpstream
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
+import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
-import io.infinitape.etherjar.rpc.Commands
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
@@ -72,8 +70,7 @@ class TrackEthereumTx(

    override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
        val base = prepareTracking(request)
-        val up = upstreams.getUpstream(base.chain)?.castApi(EthereumApi::class.java)
-                ?: return Flux.empty()
+        val up = getUpstream(base.chain)
        return update(base)
                .defaultIfEmpty(base)
                .flatMapMany {
@@ -82,9 +79,18 @@ class TrackEthereumTx(
                            .map(this@TrackEthereumTx::asProto)
                            .subscribeOn(scheduler)
                }
+                .doOnError { t ->
+                    log.error("Subscription error", t)
+                }
    }

-    fun subscribe(base: TxDetails, up: Upstream<EthereumApi>): Flux<TxDetails> {
+
+    fun getUpstream(chain: Chain): AggregatedEthereumUpstreams {
+        return upstreams.getUpstream(chain)?.cast(AggregatedEthereumUpstreams::class.java, EthereumApi::class.java)
+                ?: throw SilentException.UnsupportedBlockchain(chain)
+    }
+
+    fun subscribe(base: TxDetails, up: AggregatedEthereumUpstreams): Flux<TxDetails> {
        var latestTx = base

        val untilFound = Mono.just(latestTx)
@@ -151,11 +157,9 @@ class TrackEthereumTx(

    private fun update(tx: TxDetails): Mono<TxDetails> {
        val initialStatus = tx.status
-        val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream<EthereumApi>?
-                ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain))
-        val execution = upstream.getApi(Selector.empty)
-                .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) }
-        return execution
+        val upstream = getUpstream(tx.chain)
+        return upstream.getReader()
+                .txByHash().read(tx.txid)
                .onErrorResume(RpcException::class.java) { t ->
                    log.warn("Upstream error, ignoring. {}", t.rpcMessage)
                    Mono.empty<TransactionJson>()
@@ -198,10 +202,12 @@ class TrackEthereumTx(
    }

    private fun loadWeight(tx: TxDetails): Mono<TxDetails> {
-        val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream<EthereumApi>?
-                ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain))
-        return upstream.getApi(Selector.empty)
-                .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) }
+        val upstream = getUpstream(tx.chain)
+        if (tx.status.blockHash == null) {
+            return Mono.empty()
+        }
+        return upstream.getReader()
+                .blocksByHash().read(tx.status.blockHash)
                .map { block ->
                    setBlockDetails(tx, block)
                }.doOnError { t ->
@@ -209,7 +215,7 @@ class TrackEthereumTx(
                }
    }

-    fun updateFromBlock(upstream: Upstream<EthereumApi>, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
+    fun updateFromBlock(upstream: AggregatedEthereumUpstreams, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
        return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) {
            val updated = tx.withStatus(
                    blockHash = blockTx.blockHash,
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
@@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.upstream

 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
@@ -28,7 +28,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams
+import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
@@ -65,7 +65,7 @@ class CurrentUpstreams(
                            .cast(EthereumUpstream::class.java, EthereumApi::class.java) as Upstream<EthereumApi>
                    val current = chainMapping[chain] as ChainUpstreams<EthereumApi>?
                    val factory = Callable {
-                        EthereumChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<EthereumApi>
+                        AggregatedEthereumUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<EthereumApi>
                    }
                    processUpdate(change, up, current, factory)
                }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
@@ -21,12 +21,12 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono

-interface Upstream<out T : UpstreamApi> {
+interface Upstream<out A : UpstreamApi> {
    fun isAvailable(): Boolean
    fun getStatus(): UpstreamAvailability
    fun observeStatus(): Flux<UpstreamAvailability>
    fun getHead(): Head
-    fun getApi(matcher: Selector.Matcher): Mono<out T>
+    fun getApi(matcher: Selector.Matcher): Mono<out A>
    fun getOptions(): UpstreamsConfig.Options
    fun setLag(lag: Long)
    fun getLag(): Long
@@ -34,6 +34,6 @@ interface Upstream<out T : UpstreamApi> {
    fun getMethods(): CallMethods
    fun getId(): String

-    fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A>
+    fun <TA : UpstreamApi> castApi(apiType: Class<TA>): Upstream<TA>
    fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
@@ -19,13 +19,13 @@ import io.emeraldpay.dshackle.upstream.Head
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle

-open class BitcoinData(
+open class BitcoinReader(
        api: DirectBitcoinApi,
        head: Head
 ) : Lifecycle {

    companion object {
-        private val log = LoggerFactory.getLogger(BitcoinData::class.java)
+        private val log = LoggerFactory.getLogger(BitcoinReader::class.java)
    }

    private val mempool = CachingMempoolData(api, head)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
@@ -42,7 +42,7 @@ open class BitcoinUpstream(

    private val head: Head = createHead()
    private var validatorSubscription: Disposable? = null
-    private val data = BitcoinData(api, head)
+    private val data = BitcoinReader(api, head)

    private fun createHead(): Head {
        return BitcoinRpcHead(
@@ -51,7 +51,7 @@ open class BitcoinUpstream(
        )
    }

-    open fun getData(): BitcoinData {
+    open fun getData(): BitcoinReader {
        return data
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/AggregatedEthereumUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/AggregatedEthereumUpstreams.kt
@@ -24,7 +24,7 @@ import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle

-class EthereumChainUpstreams(
+open class AggregatedEthereumUpstreams(
        chain: Chain,
        val upstreams: MutableList<EthereumUpstream>,
        caches: Caches,
@@ -32,11 +32,13 @@ class EthereumChainUpstreams(
 ) : ChainUpstreams<EthereumApi>(chain, upstreams as MutableList<Upstream<EthereumApi>>, caches, objectMapper) {

    companion object {
-        private val log = LoggerFactory.getLogger(EthereumChainUpstreams::class.java)
+        private val log = LoggerFactory.getLogger(AggregatedEthereumUpstreams::class.java)
    }

    private var head: Head? = null

+    private val reader: EthereumReader = EthereumReader(this, this.caches, objectMapper)
+
    init {
        this.init()
    }
@@ -48,6 +50,24 @@ class EthereumChainUpstreams(
        super.init()
    }

+    override fun start() {
+        super.start()
+        reader.start()
+    }
+
+    override fun stop() {
+        super.stop()
+        reader.stop()
+    }
+
+    override fun isRunning(): Boolean {
+        return super.isRunning() || reader.isRunning
+    }
+
+    open fun getReader(): EthereumReader {
+        return reader
+    }
+
    override fun getHead(): Head {
        return head!!
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt
@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum

 import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamApi
 import io.infinitape.etherjar.rpc.*
@@ -37,6 +38,14 @@ abstract class EthereumApi(
    private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
    var upstream: Upstream<EthereumApi>? = null

+    fun <JS, RS> reader(): Reader<RpcCall<JS, RS>, RS> {
+        return object : Reader<RpcCall<JS, RS>, RS> {
+            override fun read(key: RpcCall<JS, RS>): Mono<RS> {
+                return this@EthereumApi.executeAndConvert(key)
+            }
+        }
+    }
+
    fun <JS, RS> execute(rpcCall: RpcCall<JS, RS>): Mono<ByteArray> {
        return execute(0, rpcCall.method, rpcCall.params as List<Any>)
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt
@@ -0,0 +1,172 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
+import io.emeraldpay.dshackle.cache.CurrentBlockCache
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.TxContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.reader.*
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.Upstream
+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.Commands
+import io.infinitape.etherjar.rpc.RpcCall
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.BlockTag
+import io.infinitape.etherjar.rpc.json.TransactionJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+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
+
+open class EthereumReader(
+        private val up: Upstream<EthereumApi>,
+        private val caches: Caches,
+        private val objectMapper: ObjectMapper
+) : Lifecycle {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EthereumReader::class.java)
+    }
+
+    private var headListener: Disposable? = null
+    private val balanceCache = CurrentBlockCache<Address, Wei>()
+
+    private val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block ->
+        objectMapper
+                .readValue(block.json, BlockJson::class.java)
+                .withoutTransactionDetails()
+    }
+
+    private val extractTx = Function<TxContainer, TransactionJson> { tx ->
+        objectMapper
+                .readValue(tx.json, TransactionJson::class.java)
+    }
+
+    private val blocksDirect: Reader<BlockHash, BlockJson<TransactionRefJson>>
+    private val txDirect: Reader<TransactionId, TransactionJson>
+    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) }
+
+    init {
+        blocksDirect = object : Reader<BlockHash, BlockJson<TransactionRefJson>> {
+            override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
+                return up.getApi(Selector.empty).flatMap { api ->
+                    api.executeAndConvert(Commands.eth().getBlock(key))
+                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $key")))
+                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                            .doOnNext { block ->
+                                caches.cache(Caches.Tag.REQUESTED, BlockContainer.from(block, objectMapper))
+                            }
+                }
+            }
+        }
+        txDirect = object : Reader<TransactionId, TransactionJson> {
+            override fun read(key: TransactionId): Mono<TransactionJson> {
+                return up.getApi(Selector.empty).flatMap { api ->
+                    api.executeAndConvert(Commands.eth().getTransaction(key))
+                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
+                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                            .doOnNext { tx ->
+                                if (tx.blockNumber != null && tx.blockHash != null) {
+                                    caches.cache(Caches.Tag.REQUESTED, TxContainer.from(tx, objectMapper))
+                                }
+                            }
+                }
+            }
+        }
+        balanceDirect = object : Reader<Address, Wei> {
+            override fun read(key: Address): Mono<Wei> {
+                return up.getApi(Selector.empty).flatMap { api ->
+                    api.executeAndConvert(Commands.eth().getBalance(key, BlockTag.LATEST))
+                            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
+                            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                            .doOnNext { value ->
+                                balanceCache.put(key, value)
+                            }
+                }
+            }
+        }
+    }
+
+    fun blocksById(): Reader<BlockId, BlockJson<TransactionRefJson>> {
+        return CompoundReader(
+                TransformingReader(caches.getBlocksByHash(), extractBlock),
+                RekeyingReader(idToBlockHash, blocksDirect)
+        )
+    }
+
+    fun blocksByHash(): Reader<BlockHash, BlockJson<TransactionRefJson>> {
+        return CompoundReader(
+                RekeyingReader(
+                        blockHashToId,
+                        TransformingReader(caches.getBlocksByHash(), extractBlock)
+                ),
+                blocksDirect
+        )
+    }
+
+    fun txByHash(): Reader<TransactionId, TransactionJson> {
+        return CompoundReader(
+                RekeyingReader(
+                        txHashToId,
+                        TransformingReader(caches.getTxByHash(), extractTx)
+                ),
+                txDirect
+        )
+    }
+
+    fun balance(): Reader<Address, Wei> {
+        return CompoundReader(
+                balanceCache, balanceDirect
+        )
+    }
+
+    override fun isRunning(): Boolean {
+        return this.headListener != null
+    }
+
+    override fun start() {
+        this.headListener = up.getHead().getFlux().subscribe {
+            balanceCache.evict()
+        }
+    }
+
+    override fun stop() {
+        val headListener = this.headListener
+        this.headListener = null
+        headListener?.dispose()
+    }
+}