StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #14 from emeraldpay/feat/quorum-for-direct

Browse Source 
problem: quorum is not applied for internal reads
This commit is contained in:
Igor Artamonov
2020-05-26 22:51:47 -04:00
committed by GitHub
parent 5584858e44 948c34a0c7
commit 0894b98081
10 changed files with 353 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt
View File

@@ -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> {

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt
View File

@@ -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> {

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -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

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -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()

134
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt Normal file
View File

@@ -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 }
}
}

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

@@ -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()

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

@@ -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.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.calls.CallMethods
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
)
}

174
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy Normal file
View File

@@ -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))
}
}

21
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy
View File

@@ -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:

4
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy
View File

@@ -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
)