From d551d95a3fa7ff4d39d2d217c3d5ddf5e417b20c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 7 Feb 2020 23:51:28 -0500 Subject: [PATCH] solution: cache requested transactions --- .../dshackle/cache/BlockByHeight.kt | 2 +- .../dshackle/cache/BlocksMemCache.kt | 9 +- .../io/emeraldpay/dshackle/cache/Caches.kt | 131 ++++++++++++++++ .../dshackle/cache/CachesEnabled.kt | 10 ++ .../emeraldpay/dshackle/cache/HeightCache.kt | 7 +- .../emeraldpay/dshackle/cache/TxMemCache.kt | 60 ++++++++ .../dshackle/upstream/AggregatedUpstream.kt | 14 +- .../dshackle/upstream/CachingEthereumApi.kt | 63 +++++--- .../dshackle/upstream/ChainUpstreams.kt | 4 +- .../dshackle/upstream/ConfiguredUpstreams.kt | 1 + .../dshackle/upstream/CurrentUpstreams.kt | 10 +- .../dshackle/upstream/UpstreamChange.kt | 10 +- .../upstream/ethereum/DirectEthereumApi.kt | 80 +++++++++- .../upstream/ethereum/EthereumUpstream.kt | 8 +- .../dshackle/upstream/grpc/GrpcUpstream.kt | 11 +- .../dshackle/cache/CachesSpec.groovy | 145 ++++++++++++++++++ .../dshackle/cache/TxMemCacheSpec.groovy | 131 ++++++++++++++++ .../dshackle/test/EthereumApiMock.groovy | 2 +- .../dshackle/test/EthereumApiStub.groovy | 2 +- .../dshackle/test/TestingCommons.groovy | 3 +- .../dshackle/test/UpstreamsMock.groovy | 4 +- .../upstream/AggregatedUpstreamSpec.groovy | 3 +- .../upstream/CachingEthereumApiSpec.groovy | 13 +- .../dshackle/upstream/FilteredApisSpec.groovy | 2 +- .../ethereum/DirectEthereumApiSpec.groovy | 110 ++++++++++++- 25 files changed, 775 insertions(+), 60 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/CachesEnabled.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt index b3b10681..cc04b9c5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt @@ -10,7 +10,7 @@ import reactor.core.publisher.Mono /** * Connects two caches to read through them. First is cache height->hash, second is hash->block. */ -class BlockByHeight( +open class BlockByHeight( private val heights: Reader, private val blocks: Reader> ): Reader> { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt index dd52b152..05740ed2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt @@ -24,7 +24,7 @@ import reactor.core.publisher.Mono import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentLinkedQueue -class BlocksMemCache( +open class BlocksMemCache( val maxSize: Int = 64 ): Reader> { @@ -35,7 +35,11 @@ class BlocksMemCache( return Mono.justOrEmpty(mapping[key]) } - fun add(block: BlockJson) { + open fun get(key: BlockHash): BlockJson? { + return mapping[key] + } + + open fun add(block: BlockJson) { mapping.put(block.hash, block) queue.add(block.hash) @@ -44,4 +48,5 @@ class BlocksMemCache( mapping.remove(old) } } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt new file mode 100644 index 00000000..d98b4d9f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -0,0 +1,131 @@ +package io.emeraldpay.dshackle.cache + +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory + +open class Caches( + private val blocksByHash: BlocksMemCache, + private val blocksByHeight: HeightCache, + private val txsByHash: TxMemCache +) { + + companion object { + private val log = LoggerFactory.getLogger(Caches::class.java) + + @JvmStatic + fun newBuilder(): Builder { + return Builder() + } + + @JvmStatic + fun default(): Caches { + return newBuilder().build() + } + } + + /** + * Cache data that was just requested + */ + fun cacheRequested(data: Any) { + if (data is TransactionJson) { + cache(Tag.REQUESTED, data) + } else if (data is BlockJson<*>) { + cache(Tag.REQUESTED, data as BlockJson) + } + } + + fun cache(tag: Tag, tx: TransactionJson) { + txsByHash.add(tx) + } + + fun cache(tag: Tag, block: BlockJson) { + if (tag == Tag.LATEST) { + blocksByHash.add(block) + val replaced = blocksByHeight.add(block) + //evict cached transactions if an existing block was updated + replaced?.let { replacedBlockHash -> + var evicted = false + blocksByHash.get(replacedBlockHash)?.let { block -> + txsByHash.evict(block) + evicted = true + } + if (!evicted) { + txsByHash.evict(replacedBlockHash) + } + } + } else if (tag == Tag.REQUESTED) { + // if block with transactions was requests cache only transactions + block.transactions.forEach { tx -> + if (tx is TransactionJson) { + cache(Tag.REQUESTED, tx) + } + } + } + } + + fun getBlocksByHash(): Reader> { + return blocksByHash + } + + fun getBlockHashByHeight(): Reader { + return blocksByHeight + } + + fun getBlocksByHeight(): Reader> { + return BlockByHeight(blocksByHeight, blocksByHash) + } + + fun getTxByHash(): Reader { + return txsByHash + } + + enum class Tag { + /** + * Latest data produced by blockchain + */ + LATEST, + /** + * Data requested by client + */ + REQUESTED + } + + class Builder() { + private var blocksByHash: BlocksMemCache? = null + private var blocksByHeight: HeightCache? = null + private var txsByHash: TxMemCache? = null + + fun setBlockByHash(cache: BlocksMemCache): Builder { + blocksByHash = cache + return this + } + + fun setBlockByHeight(cache: HeightCache): Builder { + blocksByHeight = cache + return this + } + + fun setTxByHash(cache: TxMemCache): Builder { + txsByHash = cache + return this + } + + fun build(): Caches { + if (blocksByHash == null) { + blocksByHash = BlocksMemCache() + } + if (blocksByHeight == null) { + blocksByHeight = HeightCache() + } + if (txsByHash == null) { + txsByHash = TxMemCache() + } + return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesEnabled.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesEnabled.kt new file mode 100644 index 00000000..782455d2 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesEnabled.kt @@ -0,0 +1,10 @@ +package io.emeraldpay.dshackle.cache + +/** + * Service is using caches + */ +interface CachesEnabled { + + fun setCaches(caches: Caches) + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt index 7aaf20c6..f453a242 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt @@ -11,7 +11,7 @@ import java.util.concurrent.ConcurrentHashMap /** * Memory cache for blocks heights, keeps mapping height->hash. */ -class HeightCache( +open class HeightCache( val maxSize: Int = 256 ): Reader { @@ -25,7 +25,8 @@ class HeightCache( return Mono.justOrEmpty(heights[key]) } - fun add(block: BlockJson) { + open fun add(block: BlockJson): BlockHash? { + val existing = heights[block.number] heights[block.number] = block.hash // evict old numbers if full @@ -34,5 +35,7 @@ class HeightCache( heights.remove(dropHeight) dropHeight++ } + + return existing } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt new file mode 100644 index 00000000..77fa9488 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt @@ -0,0 +1,60 @@ +package io.emeraldpay.dshackle.cache + +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ConcurrentLinkedQueue + +/** + * Memory cache for transactions + */ +open class TxMemCache( + // usually there is 100-150 tx per block on Ethereum, we keep data for about 32 blocks by default + private val maxSize: Int = 125 * 32 +): Reader { + + companion object { + private val log = LoggerFactory.getLogger(TxMemCache::class.java) + } + + private val mapping = ConcurrentHashMap() + private val queue = ConcurrentLinkedQueue() + + override fun read(key: TransactionId): Mono { + return Mono.justOrEmpty(mapping[key]) + } + + open fun evict(block: BlockJson) { + block.transactions.forEach { + mapping.remove(it.hash) + } + } + + open fun evict(block: BlockHash) { + val ids = mapping.filter { it.value.blockHash == block } + ids.forEach { + mapping.remove(it.key) + } + } + + open fun add(tx: TransactionJson) { + //do not cache fresh transactions + if (tx.blockHash == null || tx.blockNumber == null) { + return + } + mapping.put(tx.hash, tx) + queue.add(tx.hash) + + while (queue.size > maxSize) { + val old = queue.remove() + mapping.remove(old) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index 73b95184..81ed0164 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -16,9 +16,7 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.dshackle.cache.BlockByHeight -import io.emeraldpay.dshackle.cache.BlocksMemCache -import io.emeraldpay.dshackle.cache.HeightCache +import io.emeraldpay.dshackle.cache.* import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import org.springframework.context.Lifecycle @@ -32,12 +30,11 @@ import java.util.function.Predicate import kotlin.concurrent.withLock abstract class AggregatedUpstream( - val objectMapper: ObjectMapper + private val objectMapper: ObjectMapper, + val caches: Caches ): Upstream, Lifecycle { private var cacheSubscription: Disposable? = null - private val blockReaderByHash = BlocksMemCache() - private val blockReaderByHeight = HeightCache() var cache: CachingEthereumApi = CachingEthereumApi.empty() private val reconfigLock = ReentrantLock() private var callMethods: CallMethods? = null @@ -109,10 +106,9 @@ abstract class AggregatedUpstream( reconfigLock.withLock { cacheSubscription?.dispose() cacheSubscription = head.getFlux().subscribe { - blockReaderByHash.add(it) - blockReaderByHeight.add(it) + caches.cache(Caches.Tag.LATEST, it) } - cache = CachingEthereumApi(objectMapper, blockReaderByHash, BlockByHeight(blockReaderByHeight, blockReaderByHash), head) + cache = CachingEthereumApi(objectMapper, caches, head) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt index 7c86f4b4..bed148dc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt @@ -16,17 +16,14 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.dshackle.reader.EmptyReader -import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.upstream.ethereum.EmptyEthereumHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.hex.HexQuantity -import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.ResponseJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.math.BigInteger @@ -34,8 +31,7 @@ import java.util.function.Function open class CachingEthereumApi( private val objectMapper: ObjectMapper, - private val cache: Reader>, - private val cacheHeight: Reader>, + private val caches: Caches, private val head: EthereumHead ): EthereumApi(objectMapper) { @@ -44,10 +40,14 @@ open class CachingEthereumApi( @JvmStatic fun empty(): CachingEthereumApi { - return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyReader(), EmptyEthereumHead()) + return CachingEthereumApi(ObjectMapper(), Caches.default(), EmptyEthereumHead()) } } + private val cacheBlocks = caches.getBlocksByHash() + private val cacheHeight = caches.getBlocksByHeight() + private val cacheTx = caches.getTxByHash() + override fun execute(id: Int, method: String, params: List): Mono { return when (method) { "eth_blockNumber" -> @@ -58,33 +58,54 @@ open class CachingEthereumApi( if (params.size == 2 && (params[1] == "false" || params[1] == false)) Mono.just(params[0]) .map { BlockHash.from(it as String) } - .flatMap(cache::read) - .map(toJson(id)) - .onErrorResume { t -> - log.warn("Error during read from cache", t) - Mono.empty() - } + .flatMap(cacheBlocks::read) + .transform(converter(id)) + .transform(finalizer()) else Mono.empty() "eth_getBlockByNumber" -> if (params.size == 2 && (params[1] == "false" || params[1] == false)) Mono.just(params[0]) .map { HexQuantity.from(it as String) } - .filter { - it.value < BigInteger.valueOf(Long.MAX_VALUE) - } + .filter { it.value < BigInteger.valueOf(Long.MAX_VALUE) } .map { it.value.toLong() } .flatMap(cacheHeight::read) - .map(toJson(id)) - .onErrorResume { t -> - log.warn("Error during read from cache", t) - Mono.empty() - } + .transform(converter(id)) + .transform(finalizer()) + else Mono.empty() + "eth_getTransactionByHash" -> + if (params.size == 1) + Mono.just(params[0]) + .map { TransactionId.from(it as String) } + .flatMap(cacheTx::read) + .transform(converter(id)) + .transform(finalizer()) else Mono.empty() else -> Mono.empty() } } + /** + * Convert to JSON RPC response + */ + fun converter(id: Int): Function, out Mono> { + return Function { mono -> + mono.map(toJson(id)) + } + } + + /** + * Handle errors and other stuff + */ + fun finalizer(): Function, Mono> { + return Function { mono -> + mono.onErrorResume { t -> + log.warn("Error during read from cache", t) + Mono.empty() + } + } + } + fun toJson(id: Int): Function { return Function { data -> val resp = ResponseJson() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 8ead0d68..3e425959 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead @@ -31,8 +32,9 @@ import java.time.Duration open class ChainUpstreams ( val chain: Chain, private val upstreams: MutableList, + caches: Caches, objectMapper: ObjectMapper -) : AggregatedUpstream(objectMapper), Lifecycle { +) : AggregatedUpstream(objectMapper, caches), Lifecycle { private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index 4271023b..e0f2fa0f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -144,6 +144,7 @@ open class ConfiguredUpstreams( } rpcApi = DirectEthereumApi( rpcClient.build(), + null, objectMapper, methods ).apply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt index baf2eb01..345bad54 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt @@ -16,6 +16,8 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -50,12 +52,18 @@ class CurrentUpstreams( log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed") } else { if (current == null) { - val created = ChainUpstreams(chain, ArrayList(), objectMapper) + val created = ChainUpstreams(chain, ArrayList(), Caches.default(), objectMapper) + if (up is CachesEnabled) { + up.setCaches(created.caches) + } created.addUpstream(up) created.start() chainMapping[chain] = created chainsBus.onNext(chain) } else { + if (up is CachesEnabled) { + up.setCaches(current.caches) + } current.addUpstream(up) } if (!callTargets.containsKey(chain)) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt index 889670bc..81e36b77 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt @@ -15,17 +15,25 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.grpc.Chain class UpstreamChange( val chain: Chain, val upstream: Upstream, val type: ChangeType -) { +): CachesEnabled { enum class ChangeType { ADDED, REVALIDATED, STALE, REMOVED, } + + override fun setCaches(caches: Caches) { + if (upstream is CachesEnabled) { + upstream.setCaches(caches) + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt index 757163bf..a0a48719 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt @@ -17,18 +17,22 @@ 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.upstream.CallMethods import io.grpc.Status import io.grpc.StatusRuntimeException +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.hex.HexQuantity import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.ResponseJson import org.slf4j.LoggerFactory import reactor.core.publisher.Mono -import reactor.core.publisher.switchIfEmpty -import java.time.Duration +import java.math.BigInteger open class DirectEthereumApi( val rpcClient: ReactorRpcClient, + var caches: Caches?, private val objectMapper: ObjectMapper, val targets: CallMethods ): EthereumApi(objectMapper) { @@ -90,8 +94,78 @@ open class DirectEthereumApi( } } + /** + * Actual request to the remote endpoint + */ private fun callUpstream(method: String, params: List): Mono { - return rpcClient.execute(RpcCall.create(method, Any::class.java, params)) + return rpcClient.execute(callMapping(method, params)) .timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout"))) + .doOnNext { value -> + caches?.cacheRequested(value) + } + } + + /** + * Prepare RpcCall with data types specific for that particular requests. In general it may return a call that just + * parses JSON into Map. But the purpose of further processing and caching for some of the requests we want + * to have actual data types. + */ + fun callMapping(method: String, params: List): RpcCall { + return when { + method == "eth_getTransactionByHash" -> { + if (params.size != 1) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter") + } + val hash: TransactionId + try { + hash = TransactionId.from(params[0].toString()) + } catch (e: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id") + } + Commands.eth().getTransaction(hash) + } + method == "eth_getBlockByHash" -> { + if (params.size != 2) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters") + } + val hash: BlockHash + try { + hash = BlockHash.from(params[0].toString()) + } catch (e: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block hash") + } + val withTx = params[1].toString().toBoolean() + if (withTx) { + Commands.eth().getBlockWithTransactions(hash) + } else { + Commands.eth().getBlock(hash) + } + } + method == "eth_getBlockByNumber" -> { + if (params.size != 2) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters") + } + val number: Long + try { + val quantity = HexQuantity.from(params[0].toString()) ?: throw IllegalArgumentException() + number = quantity.value.let { + if (it < BigInteger.valueOf(Long.MAX_VALUE) && it >= BigInteger.ZERO) { + it.toLong() + } else { + throw IllegalArgumentException() + } + } + } catch (e: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number") + } + val withTx = params[1].toString().toBoolean() + if (withTx) { + Commands.eth().getBlockWithTransactions(number) + } else { + Commands.eth().getBlock(number) + } + } + else -> RpcCall.create(method, Any::class.java, params) + } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 5d132ec4..89a20ad4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -15,6 +15,8 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.grpc.Chain @@ -32,7 +34,7 @@ open class EthereumUpstream( private val options: UpstreamsConfig.Options, val node: NodeDetailsList.NodeDetails, private val targets: CallMethods -): DefaultUpstream(), Lifecycle { +): DefaultUpstream(), CachesEnabled, Lifecycle { constructor(id: String, chain: Chain, api: DirectEthereumApi): this(id, chain, api, null, UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()), @@ -48,6 +50,10 @@ open class EthereumUpstream( api.upstream = this } + override fun setCaches(caches: Caches) { + api.caches = caches; + } + override fun getId(): String { return id } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt index da4b7a11..e7e4f496 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt @@ -21,6 +21,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead @@ -53,10 +55,11 @@ open class GrpcUpstream( private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, private val rpcClient: ReactorEmeraldClient -): DefaultUpstream(), Lifecycle { +): DefaultUpstream(), CachesEnabled, Lifecycle { private var allLabels: Collection = ArrayList() private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) + private var caches: Caches? = null private val options = UpstreamsConfig.Options.getDefaults() private val nodes = AtomicReference(NodeDetailsList()) @@ -71,7 +74,7 @@ open class GrpcUpstream( val client = Selector.extractLabels(matcher)?.let { selector -> rpcClient.copyWithSelector(selector.asProto()) } ?: rpcClient - return DirectEthereumApi(client, objectMapper, targets).let { + return DirectEthereumApi(client, caches, objectMapper, targets).let { it.upstream = this it } @@ -205,4 +208,8 @@ open class GrpcUpstream( return options } + override fun setCaches(caches: Caches) { + this.caches = caches + } + } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy new file mode 100644 index 00000000..571205c6 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy @@ -0,0 +1,145 @@ +package io.emeraldpay.dshackle.cache + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import spock.lang.Specification + +class CachesSpec extends Specification { + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + + + def "Evict txes if block updated"() { + setup: + TxMemCache txCache = Mock() + HeightCache heightCache = Mock() + BlocksMemCache blocksCache = Mock() + def caches = Caches.newBuilder() + .setTxByHash(txCache) + .setBlockByHeight(heightCache) + .setBlockByHash(blocksCache) + .build() + + def block1 = new BlockJson() + block1.number = 100 + block1.hash = BlockHash.from(hash1) + + def block2 = new BlockJson() + block2.number = 100 + block2.hash = BlockHash.from(hash2) + + when: + caches.cache(Caches.Tag.LATEST, block1) + then: + 1 * blocksCache.add(block1) + 1 * heightCache.add(block1) >> null + + when: + caches.cache(Caches.Tag.LATEST, block2) + then: + 1 * blocksCache.add(block2) + 1 * heightCache.add(block2) >> block1.hash + 1 * blocksCache.get(block1.hash) >> block1 + 1 * txCache.evict(block1) + } + + def "Evict txes if block updated - when block not cached"() { + setup: + TxMemCache txCache = Mock() + HeightCache heightCache = Mock() + BlocksMemCache blocksCache = Mock() + def caches = Caches.newBuilder() + .setTxByHash(txCache) + .setBlockByHeight(heightCache) + .setBlockByHash(blocksCache) + .build() + + def block1 = new BlockJson() + block1.number = 100 + block1.hash = BlockHash.from(hash1) + + def block2 = new BlockJson() + block2.number = 100 + block2.hash = BlockHash.from(hash2) + + when: + caches.cache(Caches.Tag.LATEST, block1) + then: + 1 * blocksCache.add(block1) + 1 * heightCache.add(block1) >> null + + when: + caches.cache(Caches.Tag.LATEST, block2) + then: + 1 * blocksCache.add(block2) + 1 * heightCache.add(block2) >> block1.hash + 1 * blocksCache.get(block1.hash) >> null + 1 * txCache.evict(block1.hash) + } + + def "Do not cache txes of a requested block if it's just id"() { + setup: + TxMemCache txCache = Mock() + HeightCache heightCache = Mock() + BlocksMemCache blocksCache = Mock() + def caches = Caches.newBuilder() + .setTxByHash(txCache) + .setBlockByHeight(heightCache) + .setBlockByHash(blocksCache) + .build() + + def block = new BlockJson() + block.number = 100 + block.hash = BlockHash.from(hash1) + block.transactions = [ + new TransactionRefJson(TransactionId.from(hash1)), + new TransactionRefJson(TransactionId.from(hash2)), + ] + + when: + caches.cache(Caches.Tag.REQUESTED, block) + then: + 0 * txCache.add(_) + } + + def "Cache txes of a requested block"() { + setup: + TxMemCache txCache = Mock() + HeightCache heightCache = Mock() + BlocksMemCache blocksCache = Mock() + def caches = Caches.newBuilder() + .setTxByHash(txCache) + .setBlockByHeight(heightCache) + .setBlockByHash(blocksCache) + .build() + + def tx1 = new TransactionJson().with { + hash = TransactionId.from(hash1) + blockHash = BlockHash.from(hash1) + blockNumber = 100 + it + } + def tx2 = new TransactionJson().with { + hash = TransactionId.from(hash2) + blockHash = BlockHash.from(hash1) + blockNumber = 100 + it + } + + + def block = new BlockJson() + block.number = 100 + block.hash = BlockHash.from(hash1) + block.transactions = [tx1, tx2] + + when: + caches.cache(Caches.Tag.REQUESTED, block) + then: + 1 * txCache.add(tx1) + 1 * txCache.add(tx2) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy new file mode 100644 index 00000000..1e292b67 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy @@ -0,0 +1,131 @@ +package io.emeraldpay.dshackle.cache + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import spock.lang.Specification + +class TxMemCacheSpec extends Specification { + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" + String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" + + def "Add and read"() { + setup: + def cache = new TxMemCache() + def tx = new TransactionJson() + tx.hash = TransactionId.from(hash1) + tx.blockHash = BlockHash.from(hash1) + tx.blockNumber = 100 + + when: + cache.add(tx) + def act = cache.read(TransactionId.from(hash1)).block() + then: + act == tx + } + + def "Keeps only configured amount"() { + setup: + def cache = new TxMemCache(3) + + when: + [hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = 100 + i + tx.blockHash = BlockHash.from(hash) + tx.hash = TransactionId.from(hash) + cache.add(tx) + } + + def act1 = cache.read(TransactionId.from(hash1)).block() + def act2 = cache.read(TransactionId.from(hash2)).block() + def act3 = cache.read(TransactionId.from(hash3)).block() + def act4 = cache.read(TransactionId.from(hash4)).block() + then: + act2.hash.toHex() == hash2 + act3.hash.toHex() == hash3 + act4.hash.toHex() == hash4 + act1 == null + } + + def "Evict all by block hash"() { + setup: + def cache = new TxMemCache() + + when: + [hash1, hash2].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = 100 + tx.blockHash = BlockHash.from(hash1) + tx.hash = TransactionId.from(hash) + cache.add(tx) + } + [hash3, hash4].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = 101 + tx.blockHash = BlockHash.from(hash2) + tx.hash = TransactionId.from(hash) + cache.add(tx) + } + + cache.evict(BlockHash.from(hash1)) + + def act1 = cache.read(TransactionId.from(hash1)).block() + def act2 = cache.read(TransactionId.from(hash2)).block() + def act3 = cache.read(TransactionId.from(hash3)).block() + def act4 = cache.read(TransactionId.from(hash4)).block() + + then: + act1 == null + act2 == null + act3.hash.toHex() == hash3 + act4.hash.toHex() == hash4 + } + + def "Evict all by block data"() { + setup: + def cache = new TxMemCache() + + when: + [hash1, hash2].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = 100 + tx.blockHash = BlockHash.from(hash1) + tx.hash = TransactionId.from(hash) + cache.add(tx) + } + [hash3, hash4].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = 100 + tx.blockHash = BlockHash.from(hash2) + tx.hash = TransactionId.from(hash) + cache.add(tx) + } + + def block = new BlockJson() + block.hash = BlockHash.from(hash1) + block.number = 100 + block.transactions = [ + new TransactionRefJson(TransactionId.from(hash1)), + new TransactionRefJson(TransactionId.from(hash2)), + ] + + cache.evict(block) + + def act1 = cache.read(TransactionId.from(hash1)).block() + def act2 = cache.read(TransactionId.from(hash2)).block() + def act3 = cache.read(TransactionId.from(hash3)).block() + def act4 = cache.read(TransactionId.from(hash4)).block() + + then: + act1 == null + act2 == null + act3.hash.toHex() == hash3 + act4.hash.toHex() == hash4 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index 275a32a7..e5d9cac3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -39,7 +39,7 @@ class EthereumApiMock extends DirectEthereumApi { private ObjectMapper objectMapper EthereumApiMock(@NotNull ReactorRpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) { - super(rpcClient, objectMapper, new DirectCallMethods()) + super(rpcClient, null, objectMapper, new DirectCallMethods()) this.objectMapper = objectMapper } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy index 977303eb..7926abf5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiStub.groovy @@ -38,7 +38,7 @@ class EthereumApiStub extends DirectEthereumApi { } EthereumApiStub(String id) { - super(rpcClient, objectMapper, new DirectCallMethods()) + super(rpcClient, null, objectMapper, new DirectCallMethods()) this.id = id } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 4d8ce975..ce3f0ff7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.Version import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.module.SimpleModule +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.CallMethods import io.emeraldpay.dshackle.upstream.ChainUpstreams @@ -73,6 +74,6 @@ class TestingCommons { } static AggregatedUpstream aggregatedUpstream(EthereumUpstream up) { - return new ChainUpstreams(Chain.ETHEREUM, [up], objectMapper()) + return new ChainUpstreams(Chain.ETHEREUM, [up], Caches.default(), objectMapper()) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy index 46712da9..f86d4ecd 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy @@ -15,7 +15,7 @@ */ package io.emeraldpay.dshackle.test - +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.ChainUpstreams import io.emeraldpay.dshackle.upstream.QuorumBasedMethods @@ -40,7 +40,7 @@ class UpstreamsMock implements Upstreams { AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { if (!upstreams.containsKey(chain)) { - upstreams[chain] = new ChainUpstreams(chain, [up], TestingCommons.objectMapper()) + upstreams[chain] = new ChainUpstreams(chain, [up], Caches.default(), TestingCommons.objectMapper()) } else { upstreams[chain].addUpstream(up) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy index a9712946..f8b517f8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons @@ -28,7 +29,7 @@ class AggregatedUpstreamSpec extends Specification { setup: def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new ChainUpstreams(Chain.ETHEREUM, [up1, up2], TestingCommons.objectMapper()) + def aggr = new ChainUpstreams(Chain.ETHEREUM, [up1, up2], Caches.default(), TestingCommons.objectMapper()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy index 03f1c5e1..ff300428 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.cache.BlockByHeight import io.emeraldpay.dshackle.cache.BlocksMemCache +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.HeightCache import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.test.TestingCommons @@ -23,8 +24,7 @@ class CachingEthereumApiSpec extends Specification { def head = Mock(EthereumHead.class) def api = new CachingEthereumApi( TestingCommons.objectMapper(), - new EmptyReader>(), - new EmptyReader<>(), + Caches.default(), head ) 1 * head.getFlux() >> Flux.just(new BlockJson(number: 100)) @@ -43,8 +43,7 @@ class CachingEthereumApiSpec extends Specification { def head = Mock(EthereumHead.class) def api = new CachingEthereumApi( TestingCommons.objectMapper(), - new EmptyReader>(), - new EmptyReader<>(), + Caches.default(), head ) when: @@ -62,8 +61,7 @@ class CachingEthereumApiSpec extends Specification { def head = Mock(EthereumHead.class) def api = new CachingEthereumApi( TestingCommons.objectMapper(), - cache, - new EmptyReader<>(), + Caches.newBuilder().setBlockByHash(cache).build(), head ) cache.add(new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58"))) @@ -85,8 +83,7 @@ class CachingEthereumApiSpec extends Specification { def head = Mock(EthereumHead.class) def api = new CachingEthereumApi( TestingCommons.objectMapper(), - blocksCache, - new BlockByHeight(heightCache, blocksCache), + Caches.newBuilder().setBlockByHash(blocksCache).setBlockByHeight(heightCache).build(), head ) def block = new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 985a4230..a3ff2298 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -47,7 +47,7 @@ class FilteredApisSpec extends Specification { new EthereumUpstream( "test", Chain.ETHEREUM, - new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets), + new DirectEthereumApi(rpcClient, null, objectMapper, ethereumTargets), (EthereumWs) null, new UpstreamsConfig.Options(), new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)), diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy index 6374b569..988af747 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApiSpec.groovy @@ -20,6 +20,8 @@ import io.emeraldpay.dshackle.upstream.DirectCallMethods import io.infinitape.etherjar.rpc.ReactorRpcClient 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 reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Specification @@ -28,7 +30,7 @@ import java.time.Duration class DirectEthereumApiSpec extends Specification { - DirectEthereumApi api = new DirectEthereumApi(Stub(ReactorRpcClient), TestingCommons.objectMapper(), new DirectCallMethods()) + DirectEthereumApi api = new DirectEthereumApi(Stub(ReactorRpcClient), null, TestingCommons.objectMapper(), new DirectCallMethods()) def "Process successful result"() { setup: @@ -89,4 +91,110 @@ class DirectEthereumApiSpec extends Specification { .verify(Duration.ofSeconds(1)) } + + def "Typed mapping for block request"() { + when: + def act = api.callMapping("eth_getBlockByHash", ["0xacf5611707048efc39cabed483e420672ca1ed070f248ef6202c99994dbc6061", false]) + then: + act.jsonType == BlockJson + act.resultType == BlockJson + } + + def "Typed mapping for block request with txes"() { + when: + def act = api.callMapping("eth_getBlockByHash", ["0xacf5611707048efc39cabed483e420672ca1ed070f248ef6202c99994dbc6061", true]) + then: + act.jsonType == BlockJson + act.resultType == BlockJson + } + + def "Typed mapping for block by height request"() { + when: + def act = api.callMapping("eth_getBlockByNumber", ["0x135", false]) + then: + act.jsonType == BlockJson + act.resultType == BlockJson + } + + def "Typed mapping for block by height request with txes"() { + when: + def act = api.callMapping("eth_getBlockByNumber", ["0xacf5", true]) + then: + act.jsonType == BlockJson + act.resultType == BlockJson + } + + def "Typed mapping for tx request"() { + when: + def act = api.callMapping("eth_getTransactionByHash", ["0xacf5611707048efc39cabed483e420672ca1ed070f248ef6202c99994dbc6061"]) + then: + act.jsonType == TransactionJson + act.resultType == TransactionJson + } + + def "Errors for mapping of invalid tx request"() { + when: + api.callMapping("eth_getTransactionByHash", ["0xacf5611707048efc39cabed483e420672ca1ed070f248ef6"]) + then: + def t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getTransactionByHash", ["0xacf5611707048efc39cabed483e420672ca1ed070f248ef6202c99994dbc6061", true]) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getTransactionByHash", []) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + } + + def "Errors for mapping of invalid block request"() { + when: + api.callMapping("eth_getBlockByHash", ["0xacf5611707048efc39cabed483e420672ca1ed070f248ef6202c99994dbc6061"]) + then: + def t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getBlockByHash", ["0xacf5611707048efc39cabed48f6202c99994dbc6061", true]) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getBlockByHash", []) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + } + + def "Errors for mapping of invalid block by number request"() { + when: + api.callMapping("eth_getBlockByNumber", ["0xacf5611707048efc3248ef6202c99994dbc6061"]) + then: + def t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getBlockByNumber", ["0x", true]) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getBlockByNumber", ["-0x23", true]) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + + when: + api.callMapping("eth_getBlockByNumber", []) + then: + t = thrown(RpcException) + t.code == RpcResponseError.CODE_INVALID_METHOD_PARAMS + } }