From e6e1d0c9c87ab70f8ec89eaa9415805eff3b65fa Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 25 Jul 2023 14:08:51 +0400 Subject: [PATCH] Return upstreamId from direct reader (#257) --- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 10 +-- .../dshackle/rpc/TrackEthereumAddress.kt | 1 + .../ethereum/EthereumCachingReader.kt | 73 +++++++-------- .../upstream/ethereum/EthereumDirectReader.kt | 90 +++++++++++-------- .../upstream/ethereum/EthereumLocalReader.kt | 22 ++--- .../ethereum/subscribe/ProduceLogs.kt | 7 +- .../upstream/rpcclient/JsonRpcResponse.kt | 3 + .../ethereum/EthereumCachingReaderSpec.groovy | 18 ++-- .../ethereum/EthereumLocalReaderSpec.groovy | 24 +++-- .../ethereum/subscribe/ProduceLogsSpec.groovy | 19 ++-- 10 files changed, 145 insertions(+), 122 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index aa3bd758..39d8ab6d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -48,7 +48,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError @@ -369,13 +368,14 @@ open class NativeCall( .flatMap { api -> SpannedReader(api, tracer, LOCAL_READER) .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) - .flatMap(JsonRpcResponse::requireResult) .map { - validateResult(it, "local", ctx) + val result = it.getResult() + val upstreamId = it.providedUpstreamId ?: ctx.upstream.getId() + validateResult(result, "local", ctx) if (ctx.nonce != null) { - CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream.getId()), ctx.upstream.getId(), ctx) + CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, upstreamId), upstreamId, ctx) } else { - CallResult.ok(ctx.id, null, it, null, ctx.upstream.getId(), ctx) + CallResult.ok(ctx.id, null, result, null, upstreamId, ctx) } } }.switchIfEmpty( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt index 28359419..1001ec8f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt @@ -121,6 +121,7 @@ class TrackEthereumAddress( .balance() .read(addr.address) .timeout(Defaults.timeout) + .map { it.data } } private fun buildResponse(address: TrackedAddress): BlockchainOuterClass.AddressBalance { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt index 4f4a36ab..18e9a37c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt @@ -19,16 +19,13 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache -import io.emeraldpay.dshackle.cache.HeightByHashAdding import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HASH_READER import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HEIGHT_READER -import io.emeraldpay.dshackle.commons.CACHE_HEIGHT_BY_HASH_READER import io.emeraldpay.dshackle.commons.CACHE_RECEIPTS_READER import io.emeraldpay.dshackle.commons.CACHE_TX_BY_HASH_READER import io.emeraldpay.dshackle.commons.DIRECT_QUORUM_RPC_READER import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.SourceContainer import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.CompoundReader @@ -39,6 +36,7 @@ import io.emeraldpay.dshackle.reader.TransformingReader import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot import io.emeraldpay.etherjar.domain.Address @@ -47,8 +45,8 @@ import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.domain.Wei import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory -import org.slf4j.LoggerFactory import org.springframework.cloud.sleuth.Tracer +import reactor.core.publisher.Mono import java.util.function.Function /** @@ -57,19 +55,16 @@ import java.util.function.Function open class EthereumCachingReader( private val up: Multistream, private val caches: Caches, - private val callMethodsFactory: Factory, + callMethodsFactory: Factory, private val tracer: Tracer ) : Lifecycle { - companion object { - private val log = LoggerFactory.getLogger(EthereumCachingReader::class.java) - } - private val objectMapper: ObjectMapper = Global.objectMapper private val balanceCache = CurrentBlockCache() private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory, tracer) - val extractBlock = Function> { block -> + private val extractBlock = Function, BlockJson> { result -> + val block = result.data val existing = block.getParsed(BlockJson::class.java) if (existing != null) { existing.withoutTransactionDetails() @@ -80,12 +75,9 @@ open class EthereumCachingReader( } } - val extractTx = Function { tx -> - tx.getParsed(TransactionJsonSnapshot::class.java) ?: objectMapper.readValue(tx.json, TransactionJsonSnapshot::class.java) - } - - val asRaw = Function { tx -> - tx.json ?: ByteArray(0) + private val extractTx = Function, TransactionJsonSnapshot> { result -> + result.data.getParsed(TransactionJsonSnapshot::class.java) + ?: objectMapper.readValue(result.data.json, TransactionJsonSnapshot::class.java) } private val idToBlockHash = Function { id -> BlockHash.from(id.value) } @@ -95,16 +87,13 @@ open class EthereumCachingReader( private val idToTxHash = Function { id -> TransactionId.from(id.value) } private val blocksByIdAsCont = CompoundReader( - SpannedReader(caches.getBlocksByHash(), tracer, CACHE_BLOCK_BY_HASH_READER), + SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER), SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER) ) - private val heightByHash = - SpannedReader(HeightByHashAdding(caches, blocksByIdAsCont), tracer, CACHE_HEIGHT_BY_HASH_READER) - - fun blocksByHashAsCont(): Reader { + fun blocksByHashAsCont(): Reader> { return CompoundReader( - SpannedReader(RekeyingReader(blockHashToId, caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER), + SpannedReader(CacheWithUpstreamIdReader(RekeyingReader(blockHashToId, caches.getBlocksByHash())), tracer, CACHE_BLOCK_BY_HASH_READER), SpannedReader(directReader.blockReader, tracer, DIRECT_QUORUM_RPC_READER) ) } @@ -116,20 +105,13 @@ open class EthereumCachingReader( ) } - fun blocksByIdParsed(): Reader> { - return TransformingReader( - blocksByIdAsCont(), - extractBlock - ) - } - - open fun blocksByIdAsCont(): Reader { + open fun blocksByIdAsCont(): Reader> { return blocksByIdAsCont } - open fun blocksByHeightAsCont(): Reader { + open fun blocksByHeightAsCont(): Reader> { return CompoundReader( - SpannedReader(caches.getBlocksByHeight(), tracer, CACHE_BLOCK_BY_HEIGHT_READER), + SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHeight()), tracer, CACHE_BLOCK_BY_HEIGHT_READER), SpannedReader(directReader.blockByHeightReader, tracer, DIRECT_QUORUM_RPC_READER) ) } @@ -144,42 +126,38 @@ open class EthereumCachingReader( open fun txByHash(): Reader { return TransformingReader( CompoundReader( - RekeyingReader(txHashToId, caches.getTxByHash()), + CacheWithUpstreamIdReader(RekeyingReader(txHashToId, caches.getTxByHash())), directReader.txReader ), extractTx ) } - open fun txByHashAsCont(): Reader { + open fun txByHashAsCont(): Reader> { return CompoundReader( - SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER), + CacheWithUpstreamIdReader(SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER)), SpannedReader(RekeyingReader(idToTxHash, directReader.txReader), tracer, DIRECT_QUORUM_RPC_READER) ) } - fun balance(): Reader { + fun balance(): Reader> { // TODO include height as part of cache? return CompoundReader( - balanceCache, directReader.balanceReader + CacheWithUpstreamIdReader(balanceCache), directReader.balanceReader ) } - fun receipts(): Reader { + fun receipts(): Reader> { val requested = RekeyingReader( { txid: TxId -> TransactionId.from(txid.value) }, directReader.receiptReader ) return CompoundReader( - SpannedReader(caches.getReceipts(), tracer, CACHE_RECEIPTS_READER), + CacheWithUpstreamIdReader(SpannedReader(caches.getReceipts(), tracer, CACHE_RECEIPTS_READER)), SpannedReader(requested, tracer, DIRECT_QUORUM_RPC_READER) ) } - fun heightByHash(): Reader { - return heightByHash - } - override fun isRunning(): Boolean { // TODO should be always running? return true // up.isRunning @@ -194,4 +172,13 @@ open class EthereumCachingReader( override fun stop() { } + + private class CacheWithUpstreamIdReader( + private val reader: Reader + ) : Reader> { + override fun read(key: K): Mono> { + return reader.read(key) + .map { Result(it, null) } + } + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index 5b40b7d3..de01432b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -53,74 +53,79 @@ class EthereumDirectReader( private val objectMapper: ObjectMapper = Global.objectMapper var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() - val blockReader: Reader - val blockByHeightReader: Reader - val txReader: Reader - val balanceReader: Reader - val receiptReader: Reader + val blockReader: Reader> + val blockByHeightReader: Reader> + val txReader: Reader> + val balanceReader: Reader> + val receiptReader: Reader> init { - blockReader = object : Reader { - override fun read(key: BlockHash): Mono { + blockReader = object : Reader> { + override fun read(key: BlockHash): Mono> { val request = JsonRpcRequest("eth_getBlockByHash", listOf(key.toHex(), false)) return readBlock(request, key.toHex()) } } - blockByHeightReader = object : Reader { - override fun read(key: Long): Mono { + blockByHeightReader = object : Reader> { + override fun read(key: Long): Mono> { val heightMatcher = Selector.HeightMatcher(key) val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false)) return readBlock(request, key.toString(), heightMatcher) } } - txReader = object : Reader { - override fun read(key: TransactionId): Mono { + txReader = object : Reader> { + override fun read(key: TransactionId): Mono> { val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex())) return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key"))) - .flatMap { txbytes -> - val tx = objectMapper.readValue(txbytes, TransactionJsonSnapshot::class.java) + .flatMap { result -> + val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java) if (tx == null) { Mono.empty() } else { - Mono.just(TxContainer.from(tx, txbytes)) + Mono.just( + Result(TxContainer.from(tx, result.data), result.upstreamId) + ) } } .doOnNext { tx -> - if (tx.blockId != null) { - caches.cache(Caches.Tag.REQUESTED, tx) + if (tx.data.blockId != null) { + caches.cache(Caches.Tag.REQUESTED, tx.data) } } } } - balanceReader = object : Reader { - override fun read(key: Address): Mono { + balanceReader = object : Reader> { + override fun read(key: Address): Mono> { val height = up.getHead().getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest" val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), height)) return readWithQuorum(request) .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key"))) .map { - val str = String(it) + val str = String(it.data) // 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)) + Result( + Wei.from(str.substring(1, str.length - 1)), + it.upstreamId + ) } else { throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value") } } .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200))) .doOnNext { value -> - balanceCache.put(key, value) + balanceCache.put(key, value.data) } } } - receiptReader = object : Reader { - override fun read(key: TransactionId): Mono { + receiptReader = object : Reader> { + override fun read(key: TransactionId): Mono> { val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex())) return readWithQuorum(request) .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Receipt not read $key"))) - .flatMap { json -> - val receipt = objectMapper.readValue(json, TransactionReceiptJson::class.java) + .flatMap { result -> + val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java) if (receipt == null) { log.debug("Empty receipt for txId $key") Mono.empty() @@ -131,11 +136,13 @@ class EthereumDirectReader( txId = TxId.from(key), blockId = BlockId.from(receipt.blockHash), height = receipt.blockNumber, - json = json, + json = result.data, parsed = receipt ) ) - Mono.just(json) + Mono.just( + result + ) } } } @@ -147,27 +154,35 @@ class EthereumDirectReader( request: JsonRpcRequest, id: String, matcher: Selector.Matcher = Selector.empty - ): Mono { + ): Mono> { return readWithQuorum(request, matcher) .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id"))) .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200))) - .flatMap { blockbytes -> - val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson? + .flatMap { result -> + val block = objectMapper.readValue(result.data, BlockJson::class.java) as BlockJson? if (block == null) { - Mono.empty() + Mono.empty() } else { - Mono.just(BlockContainer.from(block, blockbytes, "unknown")) + Mono.just( + Result( + BlockContainer.from(block, result.data, "unknown"), + result.upstreamId + ) + ) } } .doOnNext { block -> - caches.cache(Caches.Tag.REQUESTED, block) + caches.cache(Caches.Tag.REQUESTED, block.data) } } /** * Read from an Upstream applying a Quorum specific for that request */ - private fun readWithQuorum(request: JsonRpcRequest, matcher: Selector.Matcher = Selector.empty): Mono { + private fun readWithQuorum( + request: JsonRpcRequest, + matcher: Selector.Matcher = Selector.empty + ): Mono> { return Mono.just(quorumReaderFactory) .map { it.create( @@ -185,7 +200,12 @@ class EthereumDirectReader( }.flatMap { it.read(request) }.map { - it.value + Result(it.value, it.resolvedBy?.getId()) } } + + data class Result( + val data: T, + val upstreamId: String? + ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt index 5ae28784..4e59a816 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt @@ -26,7 +26,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.hex.HexQuantity import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError -import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.switchIfEmpty import java.math.BigInteger @@ -45,10 +44,6 @@ class EthereumLocalReader( private val localEnabled: Boolean ) : JsonRpcReader { - companion object { - private val log = LoggerFactory.getLogger(EthereumLocalReader::class.java) - } - override fun read(key: JsonRpcRequest): Mono { if (methods.isHardcoded(key.method)) { return Mono.just(methods.executeHardcoded(key.method)) @@ -66,7 +61,7 @@ class EthereumLocalReader( } val common = commonRequests(key) if (common != null) { - return common.map { JsonRpcResponse(it, null) } + return common.map { JsonRpcResponse(it.first, null, it.second) } } return Mono.empty() } @@ -76,7 +71,7 @@ class EthereumLocalReader( * 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 commonRequests(key: JsonRpcRequest): Mono? { + fun commonRequests(key: JsonRpcRequest): Mono>? { val method = key.method val params = key.params return when { @@ -92,8 +87,8 @@ class EthereumLocalReader( } reader.txByHashAsCont() .read(hash) - .map { it.json!! } - .switchIfEmpty { Mono.just(nullValue) } + .map { it.data.json!! to it.upstreamId } + .switchIfEmpty { Mono.just(nullValue to null) } } method == "eth_getBlockByHash" -> { if (params.size != 2) { @@ -109,7 +104,7 @@ class EthereumLocalReader( if (withTx) { null } else { - reader.blocksByIdAsCont().read(hash).map { it.json!! } + reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.upstreamId } } } method == "eth_getBlockByNumber" -> { @@ -127,13 +122,14 @@ class EthereumLocalReader( } reader.receipts() .read(hash) - .switchIfEmpty { Mono.just(nullValue) } + .map { it.data to it.upstreamId } + .switchIfEmpty { Mono.just(nullValue to null) } } else -> null } } - fun getBlockByNumber(params: List): Mono? { + fun getBlockByNumber(params: List): Mono>? { if (params.size != 2 || params[0] == null || params[1] == null) { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters") } @@ -174,6 +170,6 @@ class EthereumLocalReader( } return reader.blocksByHeightAsCont() - .read(number).map { it.json!! } + .read(number).map { it.data.json!! to it.upstreamId } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt index a7c11ba6..63fb75a6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage import io.emeraldpay.etherjar.hex.HexData @@ -31,14 +32,15 @@ import reactor.kotlin.core.publisher.switchIfEmpty import java.util.concurrent.TimeUnit class ProduceLogs( - private val receipts: Reader + private val receipts: Reader> ) { companion object { private val log = LoggerFactory.getLogger(ProduceLogs::class.java) } - constructor(upstream: EthereumLikeMultistream) : this(upstream.getReader().receipts()) + constructor(upstream: EthereumLikeMultistream) : + this(upstream.getReader().receipts()) private val objectMapper = Global.objectMapper @@ -77,6 +79,7 @@ class ProduceLogs( log.warn("Cannot find receipt for tx ${update.transactionId}") Mono.empty() } + .map { it.data } .flatMapMany { jsonBytes -> // receipt could be a null, like when the original block was replaced, etc. // so just skip it as Flux.empty diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt index b6fdfbe5..0c9ed08f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt @@ -35,6 +35,9 @@ class JsonRpcResponse( constructor(result: ByteArray?, error: JsonRpcError?) : this(result, error, NumberId(0)) + constructor(result: ByteArray?, error: JsonRpcError?, resolvedBy: String?) : + this(result, error, NumberId(0), null, resolvedBy) + companion object { private val NULL_VALUE = "null".toByteArray() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index c50a0c01..53218002 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -69,7 +69,7 @@ class EthereumDirectReaderSpec extends Specification { then: StepVerifier.create(act) .expectNextMatches { block -> - block.hash.toHexWithPrefix() == hash1 + block.data.hash.toHexWithPrefix() == hash1 } .expectComplete() .verify(Duration.ofSeconds(1)) @@ -138,7 +138,7 @@ class EthereumDirectReaderSpec extends Specification { then: StepVerifier.create(act) .expectNextMatches { block -> - block.hash.toHexWithPrefix() == hash1 + block.data.hash.toHexWithPrefix() == hash1 } .expectComplete() .verify(Duration.ofSeconds(1)) @@ -174,7 +174,7 @@ class EthereumDirectReaderSpec extends Specification { then: StepVerifier.create(act) .expectNextMatches { block -> - block.hash.toHexWithPrefix() == hash1 + block.data.hash.toHexWithPrefix() == hash1 } .expectComplete() .verify(Duration.ofSeconds(1)) @@ -208,7 +208,7 @@ class EthereumDirectReaderSpec extends Specification { when: def act = reader.receiptReader.read(TransactionId.from(hash1)) .block(Duration.ofSeconds(1)) - .with { new String(it) } + .with { new String(it.data) } then: act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}' } @@ -245,7 +245,7 @@ class EthereumDirectReaderSpec extends Specification { when: def act = reader.receiptReader.read(TransactionId.from(hash1)) .block(Duration.ofSeconds(1)) - .with { new String(it) } + .with { new String(it.data) } then: act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}' } @@ -302,7 +302,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.balanceReader.read(Address.from(address1)) + def act = reader.balanceReader.read(Address.from(address1)).map {it.data} then: StepVerifier.create(act) .expectNext(Wei.from("0x100")) @@ -334,7 +334,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.balanceReader.read(Address.from(address1)) + def act = reader.balanceReader.read(Address.from(address1)).map {it.data} then: StepVerifier.create(act) .expectNext(Wei.from("0x100")) @@ -379,7 +379,7 @@ class EthereumDirectReaderSpec extends Specification { then: StepVerifier.create(act) .expectNextMatches { block -> - block.hash.toHexWithPrefix() == hash1 + block.data.hash.toHexWithPrefix() == hash1 } .expectComplete() .verify(Duration.ofSeconds(1)) @@ -422,7 +422,7 @@ class EthereumDirectReaderSpec extends Specification { then: StepVerifier.create(act) .expectNextMatches { block -> - block.hash.toHexWithPrefix() == hash1 + block.data.hash.toHexWithPrefix() == hash1 } .expectComplete() .verify(Duration.ofSeconds(1)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy index 18e64aef..7f37ce85 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy @@ -69,7 +69,9 @@ class EthereumLocalReaderSpec extends Specification { _ * blocksByIdAsCont() >> new EmptyReader<>() _ * txByHashAsCont() >> new EmptyReader<>() 1 * blocksByHeightAsCont() >> Mock(Reader) { - 1 * read(101L) >> Mono.just(TestingCommons.blockForEthereum(101L)) + 1 * read(101L) >> Mono.just( + new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(101L), null) + ) } } def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) @@ -81,8 +83,8 @@ class EthereumLocalReaderSpec extends Specification { then: act != null with(act.block()) { - it.length > 0 - with(Global.objectMapper.readValue(it, BlockJson)) { + it.first.length > 0 + with(Global.objectMapper.readValue(it.first, BlockJson)) { number == 101 } } @@ -95,7 +97,9 @@ class EthereumLocalReaderSpec extends Specification { _ * blocksByIdAsCont() >> new EmptyReader<>() _ * txByHashAsCont() >> new EmptyReader<>() 1 * blocksByHeightAsCont() >> Mock(Reader) { - 1 * read(0L) >> Mono.just(TestingCommons.blockForEthereum(0L)) + 1 * read(0L) >> Mono.just( + new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(0L), null) + ) } } def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) @@ -107,8 +111,8 @@ class EthereumLocalReaderSpec extends Specification { then: act != null with(act.block()) { - it.length > 0 - with(Global.objectMapper.readValue(it, BlockJson)) { + it.first.length > 0 + with(Global.objectMapper.readValue(it.first, BlockJson)) { number == 0 } } @@ -121,7 +125,9 @@ class EthereumLocalReaderSpec extends Specification { _ * blocksByIdAsCont() >> new EmptyReader<>() _ * txByHashAsCont() >> new EmptyReader<>() 1 * blocksByHeightAsCont() >> Mock(Reader) { - 1 * read(74735L) >> Mono.just(TestingCommons.blockForEthereum(74735L)) + 1 * read(74735L) >> Mono.just( + new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), null) + ) } } def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) @@ -133,8 +139,8 @@ class EthereumLocalReaderSpec extends Specification { then: act != null with(act.block()) { - it.length > 0 - with(Global.objectMapper.readValue(it, BlockJson)) { + it.first.length > 0 + with(Global.objectMapper.readValue(it.first, BlockJson)) { number == 74735 } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy index b5165489..416ab447 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader import reactor.core.publisher.Mono import spock.lang.Specification @@ -38,7 +39,8 @@ class ProduceLogsSpec extends Specification { ' }' def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> + Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } def producer = new ProduceLogs(receipts) def update = new ConnectBlockUpdates.Update( @@ -61,7 +63,8 @@ class ProduceLogsSpec extends Specification { String receipt = 'null' def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> + Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } def producer = new ProduceLogs(receipts) def update = new ConnectBlockUpdates.Update( @@ -107,7 +110,8 @@ class ProduceLogsSpec extends Specification { ' }' def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> + Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } def producer = new ProduceLogs(receipts) def update = new ConnectBlockUpdates.Update( @@ -153,7 +157,8 @@ class ProduceLogsSpec extends Specification { ' }' def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> + Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } def producer = new ProduceLogs(receipts) def update = new ConnectBlockUpdates.Update( @@ -264,7 +269,8 @@ class ProduceLogsSpec extends Specification { ' }' def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(receipt.getBytes()) + 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> + Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } def producer = new ProduceLogs(receipts) def update = new ConnectBlockUpdates.Update( @@ -343,7 +349,8 @@ class ProduceLogsSpec extends Specification { ' }' def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(receipt.getBytes()) + 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> + Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } def producer = new ProduceLogs(receipts) def update1 = new ConnectBlockUpdates.Update(