problem: caching is not verified
This commit is contained in:
@@ -108,6 +108,10 @@ class NativeCall(
|
|||||||
|
|
||||||
fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
|
fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
|
||||||
return fetchFromCache(ctx)
|
return fetchFromCache(ctx)
|
||||||
|
.onErrorResume { t ->
|
||||||
|
log.warn("Failed to read from cache", t);
|
||||||
|
Mono.empty()
|
||||||
|
}
|
||||||
.switchIfEmpty(
|
.switchIfEmpty(
|
||||||
Mono.just(ctx).flatMap(this::executeOnRemote)
|
Mono.just(ctx).flatMap(this::executeOnRemote)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ import io.infinitape.etherjar.hex.HexQuantity
|
|||||||
import io.infinitape.etherjar.rpc.json.BlockJson
|
import io.infinitape.etherjar.rpc.json.BlockJson
|
||||||
import io.infinitape.etherjar.rpc.json.ResponseJson
|
import io.infinitape.etherjar.rpc.json.ResponseJson
|
||||||
import io.infinitape.etherjar.rpc.json.TransactionRefJson
|
import io.infinitape.etherjar.rpc.json.TransactionRefJson
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
import reactor.core.publisher.Mono
|
import reactor.core.publisher.Mono
|
||||||
import java.util.function.Function
|
import java.util.function.Function
|
||||||
|
|
||||||
@@ -37,6 +38,8 @@ open class CachingEthereumApi(
|
|||||||
): EthereumApi(objectMapper) {
|
): EthereumApi(objectMapper) {
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
|
private val log = LoggerFactory.getLogger(CachingEthereumApi::class.java)
|
||||||
|
|
||||||
@JvmStatic
|
@JvmStatic
|
||||||
fun empty(): CachingEthereumApi {
|
fun empty(): CachingEthereumApi {
|
||||||
return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyEthereumHead())
|
return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyEthereumHead())
|
||||||
@@ -55,6 +58,10 @@ open class CachingEthereumApi(
|
|||||||
.map { BlockHash.from(it as String) }
|
.map { BlockHash.from(it as String) }
|
||||||
.flatMap(cache::read)
|
.flatMap(cache::read)
|
||||||
.map(toJson(id))
|
.map(toJson(id))
|
||||||
|
.onErrorResume { t ->
|
||||||
|
log.warn("Error during read from cache", t)
|
||||||
|
Mono.empty()
|
||||||
|
}
|
||||||
else Mono.empty()
|
else Mono.empty()
|
||||||
else ->
|
else ->
|
||||||
Mono.empty()
|
Mono.empty()
|
||||||
|
|||||||
@@ -0,0 +1,77 @@
|
|||||||
|
package io.emeraldpay.dshackle.upstream
|
||||||
|
|
||||||
|
import io.emeraldpay.dshackle.cache.BlocksMemCache
|
||||||
|
import io.emeraldpay.dshackle.reader.BlockCacheReader
|
||||||
|
import io.emeraldpay.dshackle.reader.EmptyReader
|
||||||
|
import io.emeraldpay.dshackle.test.TestingCommons
|
||||||
|
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
|
||||||
|
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.TransactionRefJson
|
||||||
|
import reactor.core.publisher.Flux
|
||||||
|
import reactor.test.StepVerifier
|
||||||
|
import spock.lang.Specification
|
||||||
|
|
||||||
|
import java.time.Duration
|
||||||
|
|
||||||
|
class CachingEthereumApiSpec extends Specification {
|
||||||
|
|
||||||
|
def "Get blockNumber from head"() {
|
||||||
|
setup:
|
||||||
|
def head = Mock(EthereumHead.class)
|
||||||
|
def api = new CachingEthereumApi(
|
||||||
|
TestingCommons.objectMapper(),
|
||||||
|
new EmptyReader<BlockHash, BlockJson<TransactionRefJson>>(),
|
||||||
|
head
|
||||||
|
)
|
||||||
|
1 * head.getFlux() >> Flux.just(new BlockJson<TransactionRefJson>(number: 100))
|
||||||
|
when:
|
||||||
|
def act = api.execute(1, "eth_blockNumber", []).map { new String(it)}
|
||||||
|
|
||||||
|
then:
|
||||||
|
StepVerifier.create(act)
|
||||||
|
.expectNext('{"jsonrpc":"2.0","id":1,"result":"0x64"}')
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(3))
|
||||||
|
}
|
||||||
|
|
||||||
|
def "Return empty if block is not cached"() {
|
||||||
|
setup:
|
||||||
|
def head = Mock(EthereumHead.class)
|
||||||
|
def api = new CachingEthereumApi(
|
||||||
|
TestingCommons.objectMapper(),
|
||||||
|
new EmptyReader<BlockHash, BlockJson<TransactionRefJson>>(),
|
||||||
|
head
|
||||||
|
)
|
||||||
|
when:
|
||||||
|
def act = api.execute(1, "eth_getBlockByHash", ["0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58", false]).map { new String(it)}
|
||||||
|
|
||||||
|
then:
|
||||||
|
StepVerifier.create(act)
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(3))
|
||||||
|
}
|
||||||
|
|
||||||
|
def "Return block when cached"() {
|
||||||
|
setup:
|
||||||
|
def cache = new BlocksMemCache();
|
||||||
|
def head = Mock(EthereumHead.class)
|
||||||
|
def api = new CachingEthereumApi(
|
||||||
|
TestingCommons.objectMapper(),
|
||||||
|
new BlockCacheReader(cache),
|
||||||
|
head
|
||||||
|
)
|
||||||
|
cache.add(new BlockJson<TransactionRefJson>(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")))
|
||||||
|
|
||||||
|
when:
|
||||||
|
def act = api.execute(1, "eth_getBlockByHash", ["0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58", false]).map { new String(it)}
|
||||||
|
|
||||||
|
then:
|
||||||
|
StepVerifier.create(act)
|
||||||
|
.expectNext('{"jsonrpc":"2.0","id":1,"result":{"number":"0x64","hash":"0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58","transactions":[],"uncles":[]}}')
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(3))
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user