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

Pass sort in direct reader (#588)

Browse Source
This commit is contained in:
KirillPamPam
2024-11-04 15:08:31 +04:00
committed by GitHub
parent 152e6c37f3
commit 243a11e6bb
10 changed files with 63 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -442,7 +442,7 @@ open class NativeCall(
return ctx.upstream.getLocalReader() return ctx.upstream.getLocalReader()
.flatMap { api -> .flatMap { api ->
SpannedReader(api, tracer, LOCAL_READER) SpannedReader(api, tracer, LOCAL_READER)
.read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, false, ctx.upstreamFilter.matcher)) .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, false, ctx.upstreamFilter))
.map { .map {
val result = it.getResult() val result = it.getResult()
val resolvedUpstreamData = it.resolvedUpstreamData.ifEmpty { val resolvedUpstreamData = it.resolvedUpstreamData.ifEmpty {
@@ -799,16 +799,16 @@ open class NativeCall(
selector: BlockchainOuterClass.Selector?, selector: BlockchainOuterClass.Selector?,
streamRequest: Boolean, streamRequest: Boolean,
): ChainRequest { ): ChainRequest {
return toChainRequest(nonce, selector, streamRequest, Selector.empty) return toChainRequest(nonce, selector, streamRequest, Selector.UpstreamFilter.default)
} }
fun toChainRequest( fun toChainRequest(
nonce: Long?, nonce: Long?,
selector: BlockchainOuterClass.Selector?, selector: BlockchainOuterClass.Selector?,
streamRequest: Boolean, streamRequest: Boolean,
matcher: Selector.Matcher, upstreamFilter: Selector.UpstreamFilter,
): ChainRequest { ): ChainRequest {
return ChainRequest(method, params, nonce, selector, streamRequest, matcher) return ChainRequest(method, params, nonce, selector, streamRequest, upstreamFilter)
} }
} }
} }

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainRequest.kt
View File

@@ -30,7 +30,7 @@ data class ChainRequest(
val nonce: Long?, val nonce: Long?,
val selector: BlockchainOuterClass.Selector?, val selector: BlockchainOuterClass.Selector?,
val isStreamed: Boolean = false, val isStreamed: Boolean = false,
val matcher: Selector.Matcher = Selector.empty, val upstreamFilter: Selector.UpstreamFilter = Selector.UpstreamFilter.default,
) { ) {
@JvmOverloads constructor( @JvmOverloads constructor(
@@ -39,14 +39,14 @@ data class ChainRequest(
nonce: Long? = null, nonce: Long? = null,
selectors: BlockchainOuterClass.Selector? = null, selectors: BlockchainOuterClass.Selector? = null,
isStreamed: Boolean = false, isStreamed: Boolean = false,
matcher: Selector.Matcher = Selector.empty, upstreamFilter: Selector.UpstreamFilter = Selector.UpstreamFilter.default,
) : this(method, params, 1, nonce, selectors, isStreamed, matcher) ) : this(method, params, 1, nonce, selectors, isStreamed, upstreamFilter)
constructor( constructor(
method: String, method: String,
params: CallParams, params: CallParams,
matcher: Selector.Matcher, upstreamFilter: Selector.UpstreamFilter,
) : this(method, params, 1, null, null, false, matcher) ) : this(method, params, 1, null, null, false, upstreamFilter)
fun toJson(): ByteArray { fun toJson(): ByteArray {
return params.toJson(id, method) return params.toJson(id, method)

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

@@ -261,6 +261,11 @@ class Selector {
val matcher: Matcher, val matcher: Matcher,
) { ) {
constructor(matcher: Matcher) : this(Sort.default, matcher) constructor(matcher: Matcher) : this(Sort.default, matcher)
companion object {
@JvmStatic
val default = UpstreamFilter(empty)
}
} }
data class MultiMatcher( data class MultiMatcher(

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt
View File

@@ -64,16 +64,16 @@ open class EthereumCachingReader(
return SpannedReader(directReader.blockByFinalizationReader, tracer, DIRECT_QUORUM_RPC_READER) return SpannedReader(directReader.blockByFinalizationReader, tracer, DIRECT_QUORUM_RPC_READER)
} }
open fun blocksByIdAsCont(matcher: Selector.Matcher): Reader<BlockId, Result<BlockContainer>> { open fun blocksByIdAsCont(upstreamFilter: Selector.UpstreamFilter): Reader<BlockId, Result<BlockContainer>> {
val idToBlockHash = Function<BlockId, Request<BlockHash>> { id -> Request(BlockHash.from(id.value), matcher) } val idToBlockHash = Function<BlockId, Request<BlockHash>> { id -> Request(BlockHash.from(id.value), upstreamFilter) }
return CompoundReader( return CompoundReader(
SpannedReader(CacheWithUpstreamIdReader(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), SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER),
) )
} }
open fun blocksByHeightAsCont(matcher: Selector.Matcher): Reader<Long, Result<BlockContainer>> { open fun blocksByHeightAsCont(upstreamFilter: Selector.UpstreamFilter): Reader<Long, Result<BlockContainer>> {
val numToRequest = Function<Long, Request<Long>> { num -> Request(num, matcher) } val numToRequest = Function<Long, Request<Long>> { num -> Request(num, upstreamFilter) }
return CompoundReader( return CompoundReader(
SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHeight()), tracer, CACHE_BLOCK_BY_HEIGHT_READER), SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHeight()), tracer, CACHE_BLOCK_BY_HEIGHT_READER),
SpannedReader(RekeyingReader(numToRequest, directReader.blockByHeightReader), tracer, DIRECT_QUORUM_RPC_READER), SpannedReader(RekeyingReader(numToRequest, directReader.blockByHeightReader), tracer, DIRECT_QUORUM_RPC_READER),
@@ -84,8 +84,8 @@ open class EthereumCachingReader(
return directReader.logsByHashReader return directReader.logsByHashReader
} }
open fun txByHashAsCont(matcher: Selector.Matcher): Reader<TxId, Result<TxContainer>> { open fun txByHashAsCont(upstreamFilter: Selector.UpstreamFilter): Reader<TxId, Result<TxContainer>> {
val idToTxHash = Function<TxId, Request<TransactionId>> { id -> Request(TransactionId.from(id.value), matcher) } val idToTxHash = Function<TxId, Request<TransactionId>> { id -> Request(TransactionId.from(id.value), upstreamFilter) }
return CompoundReader( return CompoundReader(
CacheWithUpstreamIdReader(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), SpannedReader(RekeyingReader(idToTxHash, directReader.txReader), tracer, DIRECT_QUORUM_RPC_READER),
@@ -100,9 +100,9 @@ open class EthereumCachingReader(
) )
} }
fun receipts(matcher: Selector.Matcher): Reader<TxId, Result<ByteArray>> { fun receipts(upstreamFilter: Selector.UpstreamFilter): Reader<TxId, Result<ByteArray>> {
val requested = RekeyingReader( val requested = RekeyingReader(
{ txid: TxId -> Request(TransactionId.from(txid.value), matcher) }, { txid: TxId -> Request(TransactionId.from(txid.value), upstreamFilter) },
directReader.receiptReader, directReader.receiptReader,
) )
return CompoundReader( return CompoundReader(

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
View File

@@ -73,19 +73,19 @@ class EthereumDirectReader(
blockReader = object : Reader<Request<BlockHash>, Result<BlockContainer>> { blockReader = object : Reader<Request<BlockHash>, Result<BlockContainer>> {
override fun read(key: Request<BlockHash>): Mono<Result<BlockContainer>> { override fun read(key: Request<BlockHash>): Mono<Result<BlockContainer>> {
val request = ChainRequest("eth_getBlockByHash", ListParams(key.requestBy.toHex(), false)) val request = ChainRequest("eth_getBlockByHash", ListParams(key.requestBy.toHex(), false))
return readBlock(request, key.requestBy.toHex(), key.matcher) return readBlock(request, key.requestBy.toHex(), key.upstreamFilter.matcher, key.upstreamFilter.sort)
} }
} }
blockByHeightReader = object : Reader<Request<Long>, Result<BlockContainer>> { blockByHeightReader = object : Reader<Request<Long>, Result<BlockContainer>> {
override fun read(key: Request<Long>): Mono<Result<BlockContainer>> { override fun read(key: Request<Long>): Mono<Result<BlockContainer>> {
val request = ChainRequest("eth_getBlockByNumber", ListParams(HexQuantity.from(key.requestBy).toHex(), false)) val request = ChainRequest("eth_getBlockByNumber", ListParams(HexQuantity.from(key.requestBy).toHex(), false))
return readBlock(request, key.toString(), key.matcher) return readBlock(request, key.toString(), key.upstreamFilter.matcher, key.upstreamFilter.sort)
} }
} }
txReader = object : Reader<Request<TransactionId>, Result<TxContainer>> { txReader = object : Reader<Request<TransactionId>, Result<TxContainer>> {
override fun read(key: Request<TransactionId>): Mono<Result<TxContainer>> { override fun read(key: Request<TransactionId>): Mono<Result<TxContainer>> {
val request = ChainRequest("eth_getTransactionByHash", ListParams(key.requestBy.toHex())) val request = ChainRequest("eth_getTransactionByHash", ListParams(key.requestBy.toHex()))
return readWithQuorum(request, key.matcher) // retries were removed because we use NotNullQuorum which handle errors too return readWithQuorum(request, key.upstreamFilter.matcher, key.upstreamFilter.sort) // retries were removed because we use NotNullQuorum which handle errors too
.timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Tx not read $key"))) .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Tx not read $key")))
.flatMap { result -> .flatMap { result ->
val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java) val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java)
@@ -152,7 +152,7 @@ class EthereumDirectReader(
receiptReader = object : Reader<Request<TransactionId>, Result<ByteArray>> { receiptReader = object : Reader<Request<TransactionId>, Result<ByteArray>> {
override fun read(key: Request<TransactionId>): Mono<Result<ByteArray>> { override fun read(key: Request<TransactionId>): Mono<Result<ByteArray>> {
val request = ChainRequest("eth_getTransactionReceipt", ListParams(key.requestBy.toHex())) val request = ChainRequest("eth_getTransactionReceipt", ListParams(key.requestBy.toHex()))
return readWithQuorum(request, key.matcher) return readWithQuorum(request, key.upstreamFilter.matcher, key.upstreamFilter.sort)
.timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Receipt not read $key"))) .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Receipt not read $key")))
.flatMap { result -> .flatMap { result ->
val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java) val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java)
@@ -274,6 +274,6 @@ class EthereumDirectReader(
data class Request<T>( data class Request<T>(
val requestBy: T, val requestBy: T,
val matcher: Selector.Matcher, val upstreamFilter: Selector.UpstreamFilter,
) )
} }

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

@@ -88,7 +88,7 @@ class EthereumLocalReader(
} catch (e: IllegalArgumentException) { } catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id") throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
} }
reader.txByHashAsCont(key.matcher) reader.txByHashAsCont(key.upstreamFilter)
.read(hash) .read(hash)
.map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } .map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) }
} }
@@ -107,14 +107,14 @@ class EthereumLocalReader(
if (withTx) { if (withTx) {
null null
} else { } else {
reader.blocksByIdAsCont(key.matcher).read(hash).map { reader.blocksByIdAsCont(key.upstreamFilter).read(hash).map {
ChainResponse(it.data.json, null, it.resolvedUpstreamData) ChainResponse(it.data.json, null, it.resolvedUpstreamData)
} }
} }
} }
method == "eth_getBlockByNumber" -> { method == "eth_getBlockByNumber" -> {
getBlockByNumber(params.list, key.matcher) getBlockByNumber(params.list, key.upstreamFilter)
} }
method == "eth_getTransactionReceipt" -> { method == "eth_getTransactionReceipt" -> {
@@ -127,7 +127,7 @@ class EthereumLocalReader(
} catch (e: IllegalArgumentException) { } catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id") throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
} }
reader.receipts(key.matcher) reader.receipts(key.upstreamFilter)
.read(hash) .read(hash)
.map { ChainResponse(it.data, null, it.resolvedUpstreamData) } .map { ChainResponse(it.data, null, it.resolvedUpstreamData) }
} }
@@ -142,7 +142,7 @@ class EthereumLocalReader(
return null return null
} }
fun getBlockByNumber(params: List<Any?>, matcher: Selector.Matcher): Mono<ChainResponse>? { fun getBlockByNumber(params: List<Any?>, upstreamFilter: Selector.UpstreamFilter): Mono<ChainResponse>? {
if (params.size != 2 || params[0] == null || params[1] == null) { if (params.size != 2 || params[0] == null || params[1] == null) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters") throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
} }
@@ -190,7 +190,7 @@ class EthereumLocalReader(
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be a block number") throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be a block number")
} }
return reader.blocksByHeightAsCont(matcher) return reader.blocksByHeightAsCont(upstreamFilter)
.read(number).map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } .read(number).map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) }
} }

18
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy
View File

@@ -94,7 +94,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.blockReader.read(new EthereumDirectReader.Request<BlockHash>(BlockHash.from(hash1), Selector.empty)) def act = reader.blockReader.read(new EthereumDirectReader.Request<BlockHash>(BlockHash.from(hash1), Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectNextMatches { block -> .expectNextMatches { block ->
@@ -122,7 +122,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.blockReader.read(new EthereumDirectReader.Request<BlockHash>(BlockHash.from(hash1), Selector.empty)) def act = reader.blockReader.read(new EthereumDirectReader.Request<BlockHash>(BlockHash.from(hash1), Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectComplete() .expectComplete()
@@ -155,7 +155,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.blockByHeightReader.read(new EthereumDirectReader.Request<Long>(100L, Selector.empty)) def act = reader.blockByHeightReader.read(new EthereumDirectReader.Request<Long>(100L, Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectNextMatches { block -> .expectNextMatches { block ->
@@ -220,7 +220,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.txReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.empty)) def act = reader.txReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectNextMatches { block -> .expectNextMatches { block ->
@@ -253,7 +253,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.receiptReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.empty)) def act = reader.receiptReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.UpstreamFilter.default))
.block(Duration.ofSeconds(1)) .block(Duration.ofSeconds(1))
.with { new String(it.data) } .with { new String(it.data) }
then: then:
@@ -287,7 +287,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.receiptReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.empty)) def act = reader.receiptReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.UpstreamFilter.default))
.block(Duration.ofSeconds(1)) .block(Duration.ofSeconds(1))
.with { new String(it.data) } .with { new String(it.data) }
then: then:
@@ -312,7 +312,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = reader.txReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.empty)) def act = reader.txReader.read(new EthereumDirectReader.Request<TransactionId>(TransactionId.from(hash1), Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectComplete() .expectComplete()
@@ -411,7 +411,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = ethereumDirectReader.blockReader.read(new EthereumDirectReader.Request<BlockHash>(BlockHash.from(hash1), Selector.empty)) def act = ethereumDirectReader.blockReader.read(new EthereumDirectReader.Request<BlockHash>(BlockHash.from(hash1), Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectNextMatches { block -> .expectNextMatches { block ->
@@ -451,7 +451,7 @@ class EthereumDirectReaderSpec extends Specification {
} }
} }
when: when:
def act = ethereumDirectReader.blockByHeightReader.read(new EthereumDirectReader.Request<Long>(100L, Selector.empty)) def act = ethereumDirectReader.blockByHeightReader.read(new EthereumDirectReader.Request<Long>(100L, Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
.expectNextMatches { block -> .expectNextMatches { block ->

32
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
View File

@@ -70,9 +70,9 @@ class EthereumLocalReaderSpec extends Specification {
1 * getCurrentHeight() >> 101L 1 * getCurrentHeight() >> 101L
} }
def reader = Mock(EthereumCachingReader) { def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
1 * blocksByHeightAsCont(Selector.empty) >> Mock(Reader) { 1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) {
1 * read(101L) >> Mono.just( 1 * read(101L) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(101L), List.of()) new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(101L), List.of())
) )
@@ -82,7 +82,7 @@ class EthereumLocalReaderSpec extends Specification {
def router = new EthereumLocalReader(reader, methods, head, null) def router = new EthereumLocalReader(reader, methods, head, null)
when: when:
def act = router.getBlockByNumber(["latest", false], Selector.empty) def act = router.getBlockByNumber(["latest", false], Selector.UpstreamFilter.default)
then: then:
act != null act != null
@@ -98,9 +98,9 @@ class EthereumLocalReaderSpec extends Specification {
setup: setup:
def head = Stub(Head) {} def head = Stub(Head) {}
def reader = Mock(EthereumCachingReader) { def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
1 * blocksByHeightAsCont(Selector.empty) >> Mock(Reader) { 1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) {
1 * read(0L) >> Mono.just( 1 * read(0L) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(0L), List.of()) new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(0L), List.of())
) )
@@ -110,7 +110,7 @@ class EthereumLocalReaderSpec extends Specification {
def router = new EthereumLocalReader(reader, methods, head, null) def router = new EthereumLocalReader(reader, methods, head, null)
when: when:
def act = router.getBlockByNumber(["earliest", false], Selector.empty) def act = router.getBlockByNumber(["earliest", false], Selector.UpstreamFilter.default)
then: then:
act != null act != null
@@ -126,9 +126,9 @@ class EthereumLocalReaderSpec extends Specification {
setup: setup:
def head = Stub(Head) {} def head = Stub(Head) {}
def reader = Mock(EthereumCachingReader) { def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
1 * blocksByHeightAsCont(Selector.empty) >> Mock(Reader) { 1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) {
1 * read(74735L) >> Mono.just( 1 * read(74735L) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), List.of()) new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), List.of())
) )
@@ -138,7 +138,7 @@ class EthereumLocalReaderSpec extends Specification {
def router = new EthereumLocalReader(reader, methods, head, null) def router = new EthereumLocalReader(reader, methods, head, null)
when: when:
def act = router.getBlockByNumber(["0x123ef", false], Selector.empty) def act = router.getBlockByNumber(["0x123ef", false], Selector.UpstreamFilter.default)
then: then:
act != null act != null
@@ -184,15 +184,15 @@ class EthereumLocalReaderSpec extends Specification {
setup: setup:
def head = Mock(Head) def head = Mock(Head)
def reader = Mock(EthereumCachingReader) { def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * blocksByHeightAsCont(Selector.empty) >> new EmptyReader<>() _ * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
} }
def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false)
def router = new EthereumLocalReader(reader, methods, head, null) def router = new EthereumLocalReader(reader, methods, head, null)
when: when:
def act = router.getBlockByNumber(["0x0", true], Selector.empty) def act = router.getBlockByNumber(["0x0", true], Selector.UpstreamFilter.default)
then: then:
act == null act == null

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

@@ -133,7 +133,7 @@ class GenericWsHeadSpec extends Specification {
new WsSubscriptions.SubscribeData(Flux.error(new RuntimeException()), "id", new AtomicReference<String>("")), new WsSubscriptions.SubscribeData(Flux.error(new RuntimeException()), "id", new AtomicReference<String>("")),
new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference<String>("")) new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference<String>(""))
] ]
1 * it.unsubscribe(new ChainRequest("eth_unsubscribe", new ListParams(""), 2, null, null, false, Selector.empty)) >> 1 * it.unsubscribe(new ChainRequest("eth_unsubscribe", new ListParams(""), 2, null, null, false, Selector.UpstreamFilter.default)) >>
Mono.just(new ChainResponse("".bytes, null)) Mono.just(new ChainResponse("".bytes, null))
} }
@@ -453,7 +453,7 @@ class GenericWsHeadSpec extends Specification {
1 * it.subscribe(_) >> new WsSubscriptions.SubscribeData( 1 * it.subscribe(_) >> new WsSubscriptions.SubscribeData(
Flux.error(new RuntimeException()), "id", new AtomicReference<String>(subId) Flux.error(new RuntimeException()), "id", new AtomicReference<String>(subId)
) )
1 * it.unsubscribe(new ChainRequest("eth_unsubscribe", new ListParams(subId), 2, null, null, false, Selector.empty)) >> 1 * it.unsubscribe(new ChainRequest("eth_unsubscribe", new ListParams(subId), 2, null, null, false, Selector.UpstreamFilter.default)) >>
Mono.just(new ChainResponse("".bytes, null)) Mono.just(new ChainResponse("".bytes, null))
} }

6
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy
View File

@@ -60,7 +60,7 @@ class WsConnectionImplSpec extends Specification {
when: when:
Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
def act = ws.callRpc(new ChainRequest("eth_getTransactionByHash", new ListParams(["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"]), 15, null, null, false, Selector.empty)) def act = ws.callRpc(new ChainRequest("eth_getTransactionByHash", new ListParams(["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"]), 15, null, null, false, Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
@@ -92,7 +92,7 @@ class WsConnectionImplSpec extends Specification {
when: when:
Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
def act = ws.callRpc(new ChainRequest("eth_getTransactionByHash", new ListParams(["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"]), 15, null, null, false, Selector.empty)) def act = ws.callRpc(new ChainRequest("eth_getTransactionByHash", new ListParams(["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"]), 15, null, null, false, Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)
@@ -126,7 +126,7 @@ class WsConnectionImplSpec extends Specification {
when: when:
Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
def act = ws.callRpc(new ChainRequest("eth_getTransactionByHash", new ListParams(["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"]), 15, null, null, false, Selector.empty)) def act = ws.callRpc(new ChainRequest("eth_getTransactionByHash", new ListParams(["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"]), 15, null, null, false, Selector.UpstreamFilter.default))
then: then:
StepVerifier.create(act) StepVerifier.create(act)