solution: use existing caches to respond to full block request

This commit is contained in:
Igor Artamonov
2020-02-09 00:24:41 -05:00
parent d551d95a3f
commit 3c2e61af7d
6 changed files with 395 additions and 23 deletions

View File

@@ -10,16 +10,16 @@ import reactor.core.publisher.Mono
/**
* Connects two caches to read through them. First is cache height->hash, second is hash->block.
*/
open class BlockByHeight(
open class BlockByHeight<T: TransactionRefJson>(
private val heights: Reader<Long, BlockHash>,
private val blocks: Reader<BlockHash, BlockJson<TransactionRefJson>>
): Reader<Long, BlockJson<TransactionRefJson>> {
private val blocks: Reader<BlockHash, BlockJson<T>>
): Reader<Long, BlockJson<T>> {
companion object {
private val log = LoggerFactory.getLogger(BlockByHeight::class.java)
}
override fun read(key: Long): Mono<BlockJson<TransactionRefJson>> {
override fun read(key: Long): Mono<BlockJson<T>> {
return heights.read(key)
.flatMap { blocks.read(it) }
}

View File

@@ -0,0 +1,54 @@
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
import org.slf4j.LoggerFactory
import org.springframework.beans.BeanUtils
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
/**
* Reads blocks with full transactions details. Based on data contained in cashes for blocks
* and transactions, i.e. two separate caches that must be provided.
*
* If source block, with just transaction hashes is not available, it returns empty
* If any of the expected block transactions is not available it returns empty
*/
class BlocksWithTxCache(
private val blocks: BlocksMemCache,
private val txes: TxMemCache
): Reader<BlockHash, BlockJson<TransactionJson>> {
companion object {
private val log = LoggerFactory.getLogger(BlocksWithTxCache::class.java)
}
override fun read(key: BlockHash): Mono<BlockJson<TransactionJson>> {
return blocks.read(key).flatMap { block ->
if (block.transactions == null || block.transactions.isEmpty()) {
// in fact it's not necessary to create a copy, made just for code clarity but may be performance loss
val fullBlock = BlockJson<TransactionJson>()
BeanUtils.copyProperties(block, fullBlock)
Mono.just(fullBlock)
} else {
Flux.fromIterable(block.transactions)
.map { it.hash }
.flatMap { txes.read(it) }
.collectList()
.flatMap { list ->
if (block.transactions.size != list.size) {
Mono.empty<BlockJson<TransactionJson>>()
} else {
val fullBlock = BlockJson<TransactionJson>()
BeanUtils.copyProperties(block, fullBlock)
fullBlock.transactions = list
Mono.just(fullBlock)
}
}
}
}
}
}

View File

@@ -84,6 +84,14 @@ open class Caches(
return txsByHash
}
fun getFullBlocks(): Reader<BlockHash, BlockJson<TransactionJson>> {
return BlocksWithTxCache(blocksByHash, txsByHash)
}
fun getFullBlocksByHeight(): Reader<Long, BlockJson<TransactionJson>> {
return BlockByHeight(blocksByHeight, BlocksWithTxCache(blocksByHash, txsByHash))
}
enum class Tag {
/**
* Latest data produced by blockchain

View File

@@ -45,8 +45,46 @@ open class CachingEthereumApi(
}
private val cacheBlocks = caches.getBlocksByHash()
private val cacheHeight = caches.getBlocksByHeight()
private val cacheBlocksByHeight = caches.getBlocksByHeight()
private val cacheTx = caches.getTxByHash()
private val cacheFullBlocks = caches.getFullBlocks()
private val cacheFullBlocksByHeight = caches.getFullBlocksByHeight()
fun readBlockByHash(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
return if (params.size == 2) {
val includeTransactions = params[1].toString().toBoolean()
val cache = if (includeTransactions) {
cacheFullBlocks
} else {
cacheBlocks
}
Mono.just(params[0])
.map { BlockHash.from(it as String) }
.flatMap(cache::read)
.transform(converter(id))
.transform(finalizer())
}
else Mono.empty()
}
fun readBlockByNumber(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
return if (params.size == 2) {
val includeTransactions = params[1].toString().toBoolean()
val cache = if (includeTransactions) {
cacheFullBlocksByHeight
} else {
cacheBlocksByHeight
}
Mono.just(params[0])
.map { HexQuantity.from(it as String) }
.filter { it.value < BigInteger.valueOf(Long.MAX_VALUE) }
.map { it.value.toLong() }
.flatMap(cache::read)
.transform(converter(id))
.transform(finalizer())
}
else Mono.empty()
}
override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
return when (method) {
@@ -54,24 +92,8 @@ open class CachingEthereumApi(
head.getFlux().next()
.map { HexQuantity.from(it.number).toHex() }
.map(toJson(id))
"eth_getBlockByHash" ->
if (params.size == 2 && (params[1] == "false" || params[1] == false))
Mono.just(params[0])
.map { BlockHash.from(it as String) }
.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) }
.map { it.value.toLong() }
.flatMap(cacheHeight::read)
.transform(converter(id))
.transform(finalizer())
else Mono.empty()
"eth_getBlockByHash" -> readBlockByHash(id, method, params)
"eth_getBlockByNumber" -> readBlockByNumber(id, method, params)
"eth_getTransactionByHash" ->
if (params.size == 1)
Mono.just(params[0])