diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 358e48d2..a5f73f62 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -442,7 +442,7 @@ open class NativeCall( return ctx.upstream.getLocalReader() .flatMap { api -> 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 { val result = it.getResult() val resolvedUpstreamData = it.resolvedUpstreamData.ifEmpty { @@ -799,16 +799,16 @@ open class NativeCall( selector: BlockchainOuterClass.Selector?, streamRequest: Boolean, ): ChainRequest { - return toChainRequest(nonce, selector, streamRequest, Selector.empty) + return toChainRequest(nonce, selector, streamRequest, Selector.UpstreamFilter.default) } fun toChainRequest( nonce: Long?, selector: BlockchainOuterClass.Selector?, streamRequest: Boolean, - matcher: Selector.Matcher, + upstreamFilter: Selector.UpstreamFilter, ): ChainRequest { - return ChainRequest(method, params, nonce, selector, streamRequest, matcher) + return ChainRequest(method, params, nonce, selector, streamRequest, upstreamFilter) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainRequest.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainRequest.kt index a7b9cc59..266b9315 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainRequest.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainRequest.kt @@ -30,7 +30,7 @@ data class ChainRequest( val nonce: Long?, val selector: BlockchainOuterClass.Selector?, val isStreamed: Boolean = false, - val matcher: Selector.Matcher = Selector.empty, + val upstreamFilter: Selector.UpstreamFilter = Selector.UpstreamFilter.default, ) { @JvmOverloads constructor( @@ -39,14 +39,14 @@ data class ChainRequest( nonce: Long? = null, selectors: BlockchainOuterClass.Selector? = null, isStreamed: Boolean = false, - matcher: Selector.Matcher = Selector.empty, - ) : this(method, params, 1, nonce, selectors, isStreamed, matcher) + upstreamFilter: Selector.UpstreamFilter = Selector.UpstreamFilter.default, + ) : this(method, params, 1, nonce, selectors, isStreamed, upstreamFilter) constructor( method: String, params: CallParams, - matcher: Selector.Matcher, - ) : this(method, params, 1, null, null, false, matcher) + upstreamFilter: Selector.UpstreamFilter, + ) : this(method, params, 1, null, null, false, upstreamFilter) fun toJson(): ByteArray { return params.toJson(id, method) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index 9ae86539..19e77d16 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -261,6 +261,11 @@ class Selector { val matcher: Matcher, ) { constructor(matcher: Matcher) : this(Sort.default, matcher) + + companion object { + @JvmStatic + val default = UpstreamFilter(empty) + } } data class MultiMatcher( 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 8dece420..dbb16b8c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt @@ -64,16 +64,16 @@ open class EthereumCachingReader( return SpannedReader(directReader.blockByFinalizationReader, tracer, DIRECT_QUORUM_RPC_READER) } - open fun blocksByIdAsCont(matcher: Selector.Matcher): Reader> { - val idToBlockHash = Function> { id -> Request(BlockHash.from(id.value), matcher) } + open fun blocksByIdAsCont(upstreamFilter: Selector.UpstreamFilter): Reader> { + val idToBlockHash = Function> { id -> Request(BlockHash.from(id.value), upstreamFilter) } return CompoundReader( SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER), SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER), ) } - open fun blocksByHeightAsCont(matcher: Selector.Matcher): Reader> { - val numToRequest = Function> { num -> Request(num, matcher) } + open fun blocksByHeightAsCont(upstreamFilter: Selector.UpstreamFilter): Reader> { + val numToRequest = Function> { num -> Request(num, upstreamFilter) } return CompoundReader( SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHeight()), tracer, CACHE_BLOCK_BY_HEIGHT_READER), SpannedReader(RekeyingReader(numToRequest, directReader.blockByHeightReader), tracer, DIRECT_QUORUM_RPC_READER), @@ -84,8 +84,8 @@ open class EthereumCachingReader( return directReader.logsByHashReader } - open fun txByHashAsCont(matcher: Selector.Matcher): Reader> { - val idToTxHash = Function> { id -> Request(TransactionId.from(id.value), matcher) } + open fun txByHashAsCont(upstreamFilter: Selector.UpstreamFilter): Reader> { + val idToTxHash = Function> { id -> Request(TransactionId.from(id.value), upstreamFilter) } return CompoundReader( CacheWithUpstreamIdReader(SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER)), SpannedReader(RekeyingReader(idToTxHash, directReader.txReader), tracer, DIRECT_QUORUM_RPC_READER), @@ -100,9 +100,9 @@ open class EthereumCachingReader( ) } - fun receipts(matcher: Selector.Matcher): Reader> { + fun receipts(upstreamFilter: Selector.UpstreamFilter): Reader> { val requested = RekeyingReader( - { txid: TxId -> Request(TransactionId.from(txid.value), matcher) }, + { txid: TxId -> Request(TransactionId.from(txid.value), upstreamFilter) }, directReader.receiptReader, ) return CompoundReader( 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 68e03a05..a2e00178 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -73,19 +73,19 @@ class EthereumDirectReader( blockReader = object : Reader, Result> { override fun read(key: Request): Mono> { 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, Result> { override fun read(key: Request): Mono> { 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, Result> { override fun read(key: Request): Mono> { 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"))) .flatMap { result -> val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java) @@ -152,7 +152,7 @@ class EthereumDirectReader( receiptReader = object : Reader, Result> { override fun read(key: Request): Mono> { 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"))) .flatMap { result -> val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java) @@ -274,6 +274,6 @@ class EthereumDirectReader( data class Request( val requestBy: T, - val matcher: Selector.Matcher, + val upstreamFilter: Selector.UpstreamFilter, ) } 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 d73ad946..cabab516 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt @@ -88,7 +88,7 @@ class EthereumLocalReader( } catch (e: IllegalArgumentException) { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id") } - reader.txByHashAsCont(key.matcher) + reader.txByHashAsCont(key.upstreamFilter) .read(hash) .map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } } @@ -107,14 +107,14 @@ class EthereumLocalReader( if (withTx) { null } else { - reader.blocksByIdAsCont(key.matcher).read(hash).map { + reader.blocksByIdAsCont(key.upstreamFilter).read(hash).map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } } } method == "eth_getBlockByNumber" -> { - getBlockByNumber(params.list, key.matcher) + getBlockByNumber(params.list, key.upstreamFilter) } method == "eth_getTransactionReceipt" -> { @@ -127,7 +127,7 @@ class EthereumLocalReader( } catch (e: IllegalArgumentException) { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id") } - reader.receipts(key.matcher) + reader.receipts(key.upstreamFilter) .read(hash) .map { ChainResponse(it.data, null, it.resolvedUpstreamData) } } @@ -142,7 +142,7 @@ class EthereumLocalReader( return null } - fun getBlockByNumber(params: List, matcher: Selector.Matcher): Mono? { + fun getBlockByNumber(params: List, upstreamFilter: Selector.UpstreamFilter): Mono? { if (params.size != 2 || params[0] == null || params[1] == null) { 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") } - return reader.blocksByHeightAsCont(matcher) + return reader.blocksByHeightAsCont(upstreamFilter) .read(number).map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } } 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 9de037b0..157e1cee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -94,7 +94,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.blockReader.read(new EthereumDirectReader.Request(BlockHash.from(hash1), Selector.empty)) + def act = reader.blockReader.read(new EthereumDirectReader.Request(BlockHash.from(hash1), Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectNextMatches { block -> @@ -122,7 +122,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.blockReader.read(new EthereumDirectReader.Request(BlockHash.from(hash1), Selector.empty)) + def act = reader.blockReader.read(new EthereumDirectReader.Request(BlockHash.from(hash1), Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectComplete() @@ -155,7 +155,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.blockByHeightReader.read(new EthereumDirectReader.Request(100L, Selector.empty)) + def act = reader.blockByHeightReader.read(new EthereumDirectReader.Request(100L, Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectNextMatches { block -> @@ -220,7 +220,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.txReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.empty)) + def act = reader.txReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectNextMatches { block -> @@ -253,7 +253,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.receiptReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.empty)) + def act = reader.receiptReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.UpstreamFilter.default)) .block(Duration.ofSeconds(1)) .with { new String(it.data) } then: @@ -287,7 +287,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.receiptReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.empty)) + def act = reader.receiptReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.UpstreamFilter.default)) .block(Duration.ofSeconds(1)) .with { new String(it.data) } then: @@ -312,7 +312,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = reader.txReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.empty)) + def act = reader.txReader.read(new EthereumDirectReader.Request(TransactionId.from(hash1), Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectComplete() @@ -411,7 +411,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = ethereumDirectReader.blockReader.read(new EthereumDirectReader.Request(BlockHash.from(hash1), Selector.empty)) + def act = ethereumDirectReader.blockReader.read(new EthereumDirectReader.Request(BlockHash.from(hash1), Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectNextMatches { block -> @@ -451,7 +451,7 @@ class EthereumDirectReaderSpec extends Specification { } } when: - def act = ethereumDirectReader.blockByHeightReader.read(new EthereumDirectReader.Request(100L, Selector.empty)) + def act = ethereumDirectReader.blockByHeightReader.read(new EthereumDirectReader.Request(100L, Selector.UpstreamFilter.default)) then: StepVerifier.create(act) .expectNextMatches { block -> 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 54b228a8..1c36539c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy @@ -70,9 +70,9 @@ class EthereumLocalReaderSpec extends Specification { 1 * getCurrentHeight() >> 101L } def reader = Mock(EthereumCachingReader) { - _ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() - _ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() - 1 * blocksByHeightAsCont(Selector.empty) >> Mock(Reader) { + _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + 1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) { 1 * read(101L) >> Mono.just( new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(101L), List.of()) ) @@ -82,7 +82,7 @@ class EthereumLocalReaderSpec extends Specification { def router = new EthereumLocalReader(reader, methods, head, null) when: - def act = router.getBlockByNumber(["latest", false], Selector.empty) + def act = router.getBlockByNumber(["latest", false], Selector.UpstreamFilter.default) then: act != null @@ -98,9 +98,9 @@ class EthereumLocalReaderSpec extends Specification { setup: def head = Stub(Head) {} def reader = Mock(EthereumCachingReader) { - _ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() - _ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() - 1 * blocksByHeightAsCont(Selector.empty) >> Mock(Reader) { + _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + 1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) { 1 * read(0L) >> Mono.just( new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(0L), List.of()) ) @@ -110,7 +110,7 @@ class EthereumLocalReaderSpec extends Specification { def router = new EthereumLocalReader(reader, methods, head, null) when: - def act = router.getBlockByNumber(["earliest", false], Selector.empty) + def act = router.getBlockByNumber(["earliest", false], Selector.UpstreamFilter.default) then: act != null @@ -126,9 +126,9 @@ class EthereumLocalReaderSpec extends Specification { setup: def head = Stub(Head) {} def reader = Mock(EthereumCachingReader) { - _ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() - _ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() - 1 * blocksByHeightAsCont(Selector.empty) >> Mock(Reader) { + _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + 1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) { 1 * read(74735L) >> Mono.just( new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), List.of()) ) @@ -138,7 +138,7 @@ class EthereumLocalReaderSpec extends Specification { def router = new EthereumLocalReader(reader, methods, head, null) when: - def act = router.getBlockByNumber(["0x123ef", false], Selector.empty) + def act = router.getBlockByNumber(["0x123ef", false], Selector.UpstreamFilter.default) then: act != null @@ -184,15 +184,15 @@ class EthereumLocalReaderSpec extends Specification { setup: def head = Mock(Head) def reader = Mock(EthereumCachingReader) { - _ * blocksByIdAsCont(Selector.empty) >> new EmptyReader<>() - _ * txByHashAsCont(Selector.empty) >> new EmptyReader<>() - _ * blocksByHeightAsCont(Selector.empty) >> new EmptyReader<>() + _ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + _ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() + _ * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>() } def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) def router = new EthereumLocalReader(reader, methods, head, null) when: - def act = router.getBlockByNumber(["0x0", true], Selector.empty) + def act = router.getBlockByNumber(["0x0", true], Selector.UpstreamFilter.default) then: act == null diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy index c01a084e..e28c2d6e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy @@ -133,7 +133,7 @@ class GenericWsHeadSpec extends Specification { new WsSubscriptions.SubscribeData(Flux.error(new RuntimeException()), "id", new AtomicReference("")), new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id", new AtomicReference("")) ] - 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)) } @@ -453,7 +453,7 @@ class GenericWsHeadSpec extends Specification { 1 * it.subscribe(_) >> new WsSubscriptions.SubscribeData( Flux.error(new RuntimeException()), "id", new AtomicReference(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)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy index 5ba90d3b..eeff55ee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy @@ -60,7 +60,7 @@ class WsConnectionImplSpec extends Specification { when: 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: StepVerifier.create(act) @@ -92,7 +92,7 @@ class WsConnectionImplSpec extends Specification { when: 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: StepVerifier.create(act) @@ -126,7 +126,7 @@ class WsConnectionImplSpec extends Specification { when: 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: StepVerifier.create(act)