From dca0e6614f4b0c49762876e458500834a20d112d Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 21 Apr 2023 18:06:24 +0800 Subject: [PATCH] use parallel scheduler in tests for validate blocking operations --- .../dshackle/rpc/TrackEthereumTxSpec.groovy | 8 ++++---- .../startup/ConfiguredUpstreamsSpec.groovy | 20 +++++++++---------- .../test/MultistreamHolderMock.groovy | 4 ++-- .../dshackle/test/TestingCommons.groovy | 6 +++--- .../dshackle/upstream/AbstractHeadSpec.groovy | 2 +- .../dshackle/upstream/FilteredApisSpec.groovy | 4 ++-- .../upstream/HeadLagObserverSpec.groovy | 2 +- .../dshackle/upstream/MergedHeadSpec.groovy | 6 +++--- .../dshackle/upstream/MultistreamSpec.groovy | 6 +++--- .../bitcoin/BitcoinRpcHeadSpec.groovy | 2 +- .../ethereum/DefaultEthereumHeadSpec.groovy | 2 +- .../EthereumEgressSubscriptionSpec.groovy | 16 +++++++-------- .../ethereum/EthereumWsHeadSpec.groovy | 10 +++++----- .../subscribe/ConnectBlockUpdatesSpec.groovy | 10 +++++----- .../ethereum/subscribe/ConnectLogsSpec.groovy | 8 ++++---- .../subscribe/ConnectNewHeadsSpec.groovy | 2 +- .../grpc/EthereumGrpcUpstreamSpec.groovy | 6 +++--- .../upstream/grpc/GrpcHeadSpec.groovy | 4 ++-- 18 files changed, 59 insertions(+), 59 deletions(-) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index d958891c..b6039f23 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy @@ -102,7 +102,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel()) apiMock.answer("eth_getTransactionByHash", [txId], txJson) apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson) @@ -195,7 +195,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel()) def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def block = new BlockContainer( @@ -217,7 +217,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel()) def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def block = new BlockContainer( @@ -296,7 +296,7 @@ class TrackEthereumTxSpec extends Specification { } MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, multi) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.parallel()) apiMock.answerOnce("eth_getTransactionByHash", [txId], null) apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted) diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index ce70a395..69b31536 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -30,9 +30,9 @@ class ConfiguredUpstreamsSpec extends Specification { Executors.newFixedThreadPool(1), ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), + Schedulers.parallel(), null, - Schedulers.boundedElastic(), + Schedulers.parallel(), ) def methods = new UpstreamsConfig.Methods( [ @@ -62,9 +62,9 @@ class ConfiguredUpstreamsSpec extends Specification { Executors.newFixedThreadPool(1), ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), + Schedulers.parallel(), null, - Schedulers.boundedElastic(), + Schedulers.parallel(), ) def methods = new UpstreamsConfig.Methods( [ @@ -93,9 +93,9 @@ class ConfiguredUpstreamsSpec extends Specification { Executors.newFixedThreadPool(1), ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), + Schedulers.parallel(), null, - Schedulers.boundedElastic(), + Schedulers.parallel(), ) expect: configurer.getHash(node, src) == expected @@ -119,9 +119,9 @@ class ConfiguredUpstreamsSpec extends Specification { Executors.newFixedThreadPool(1), ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), + Schedulers.parallel(), null, - Schedulers.boundedElastic(), + Schedulers.parallel(), ) when: def h1 = configurer.getHash(null, "hohoho") @@ -150,9 +150,9 @@ class ConfiguredUpstreamsSpec extends Specification { Executors.newFixedThreadPool(1), ChainsConfig.default(), GrpcTracing.create(Tracing.newBuilder().build()), - Schedulers.boundedElastic(), + Schedulers.parallel(), null, - Schedulers.boundedElastic(), + Schedulers.parallel(), ) def methodsGroup = new UpstreamsConfig.MethodGroups( ["filter"] as Set, diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 797c5299..3d2d5b13 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -49,7 +49,7 @@ class MultistreamHolderMock implements MultistreamHolder { } else if (up instanceof EthereumPosRpcUpstream) { upstreams[chain] = new EthereumPosMultiStream( chain, [up as EthereumPosRpcUpstream], Caches.default(), - Schedulers.boundedElastic(), TestingCommons.tracerMock() + Schedulers.parallel(), TestingCommons.tracerMock() ) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") @@ -98,7 +98,7 @@ class MultistreamHolderMock implements MultistreamHolder { Head customHead = null EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, Schedulers.boundedElastic(), new BraveTracer(null, null, null)) + super(chain, upstreams, caches, Schedulers.parallel(), new BraveTracer(null, null, null)) } EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 8a4bac7b..33a67133 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -88,7 +88,7 @@ class TestingCommons { } static Multistream multistream(EthereumPosRpcUpstreamMock up) { - return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), Schedulers.boundedElastic(), tracerMock()).tap { + return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), Schedulers.parallel(), tracerMock()).tap { start() } } @@ -109,11 +109,11 @@ class TestingCommons { } static Multistream multistreamWithoutUpstreams(Chain chain) { - return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), tracerMock()) + return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.parallel(), tracerMock()) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { - return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(), tracerMock()) + return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.parallel(), tracerMock()) } static FileResolver fileResolver() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index aa980a7a..3aad3e43 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy @@ -138,7 +138,7 @@ class AbstractHeadSpec extends Specification { BlockContainer getHead() { return null } - }, Schedulers.boundedElastic(), new BlockValidator.AlwaysValid() , 100_000) + }, Schedulers.parallel(), new BlockValidator.AlwaysValid() , 100_000) } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 93bb7ea1..a1e07b37 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -56,8 +56,8 @@ class FilteredApisSpec extends Specification { httpFactory, new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, - Schedulers.boundedElastic(), - Schedulers.boundedElastic() + Schedulers.parallel(), + Schedulers.parallel() ) new EthereumRpcUpstream( "test", diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy index 2abf7a0c..b21bda45 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy @@ -117,7 +117,7 @@ class HeadLagObserverSpec extends Specification { TestHeadLagObserver(@NotNull Head master, @NotNull Collection followers) { super(master, followers, DistanceExtractor.@Companion::extractPowDistance, - Schedulers.boundedElastic(), Duration.ofNanos(1)) + Schedulers.parallel(), Duration.ofNanos(1)) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy index 3d84b519..cbb29802 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy @@ -37,7 +37,7 @@ class MergedHeadSpec extends Specification { } when: - def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice(), Schedulers.boundedElastic()) + def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice(), Schedulers.parallel()) merged.start() then: @@ -46,14 +46,14 @@ class MergedHeadSpec extends Specification { class TestHead1 extends AbstractHead { TestHead1() { - super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000) + super(new MostWorkForkChoice(), Schedulers.parallel(), new BlockValidator.AlwaysValid(), 100_000) } } class TestHead2 extends AbstractHead implements Lifecycle { TestHead2() { - super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000) + super(new MostWorkForkChoice(), Schedulers.parallel(), new BlockValidator.AlwaysValid(), 100_000) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 2d9680f6..a6d019c4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -51,7 +51,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock()) + def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() @@ -187,7 +187,7 @@ class MultistreamSpec extends Specification { def up1 = TestingCommons.upstream("test-1", "internal") def up2 = TestingCommons.upstream("test-2", "external") def up3 = TestingCommons.upstream("test-3", "external") - def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), Schedulers.boundedElastic(), TestingCommons.tracerMock()) + def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) expect: multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) @@ -255,7 +255,7 @@ class MultistreamSpec extends Specification { class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches, Schedulers.boundedElastic(), TestingCommons.tracerMock()) + super(chain, upstreams, caches, Schedulers.parallel(), TestingCommons.tracerMock()) } @NotNull diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy index 45a1cebf..c0ecccc1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy @@ -86,7 +86,7 @@ class BitcoinRpcHeadSpec extends Specification { _ * read(new JsonRpcRequest("getblock", [hash1])) >> Mono.just(new JsonRpcResponse(block1.bytes, null)) _ * read(new JsonRpcRequest("getblock", [hash2])) >> Mono.just(new JsonRpcResponse(block2.bytes, null)) } - BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200), Schedulers.boundedElastic()) + BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200), Schedulers.parallel()) when: def act = head.flux.take(2) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy index e7cc8a9d..d9afc122 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy @@ -32,7 +32,7 @@ import java.time.Instant class DefaultEthereumHeadSpec extends Specification { - DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic()) + DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.parallel()) ObjectMapper objectMapper = Global.objectMapper BlockHash parent = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915210") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy index 0a384473..59bc7a5a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy @@ -26,7 +26,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read empty logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([:]) @@ -37,7 +37,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read single address logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: "0x829bd824b016326a401d083b33d092293333a830" @@ -62,7 +62,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "ignores invalid address for logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: "829bd824b016326a401d083b33d092293333a830" @@ -75,7 +75,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read multi address logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: ["0x829bd824b016326a401d083b33d092293333a830", "0x401d083b33d092293333a83829bd824b016326a0"] @@ -91,7 +91,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read single topic logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ topics: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" @@ -116,7 +116,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read invalid topic for request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ topics: [ @@ -134,7 +134,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read multi topic logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ topics: [ @@ -153,7 +153,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read full logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695", diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy index aa0d3503..065cd656 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy @@ -66,7 +66,7 @@ class EthereumWsHeadSpec extends Specification { 1 * it.connectionInfoFlux() >> Flux.empty() } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.boundedElastic(), Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.parallel(), Schedulers.parallel()) when: def act = head.listenNewHeads().blockFirst() @@ -107,7 +107,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel()) when: def act = head.getFlux() @@ -161,7 +161,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel()) when: def act = head.getFlux() @@ -201,7 +201,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel()) when: def act = head.getFlux() @@ -240,7 +240,7 @@ class EthereumWsHeadSpec extends Specification { ] } - def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic()) + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel()) when: def act = head.getFlux() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy index fa343cf2..0a2e8798 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy @@ -38,7 +38,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Extracts updates"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel()) def block = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -72,7 +72,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Produce DROP updates for replaced block"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel()) def block = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -106,7 +106,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Gets prev version if available"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel()) def block1 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -170,7 +170,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Marks old txes as dropped before producing a new version of same block"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.parallel()) def block1 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -235,7 +235,7 @@ class ConnectBlockUpdatesSpec extends Specification { def up = Mock(EthereumMultistream) { 1 * getHead(Selector.empty) >> head } - def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.boundedElastic()) + def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.parallel()) when: def a1 = connectBlockUpdates.connect() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy index 0de35e85..208e6d3c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy @@ -91,7 +91,7 @@ class ConnectLogsSpec extends Specification { def "Filter is empty"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel()) when: def input = Flux.fromIterable([ log1, log2, log3, log4 @@ -109,7 +109,7 @@ class ConnectLogsSpec extends Specification { def "Filter by address"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel()) when: def input = Flux.fromIterable([ log1, log2 @@ -124,7 +124,7 @@ class ConnectLogsSpec extends Specification { def "Filter by topic"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel()) when: def input = Flux.fromIterable([ log1, log2, log3, log4 @@ -141,7 +141,7 @@ class ConnectLogsSpec extends Specification { def "Filter by address and topic"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.parallel()) when: def input = Flux.fromIterable([ log1, log2, log3, log4 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy index 998640c3..bda68a55 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy @@ -21,7 +21,7 @@ class ConnectNewHeadsSpec extends Specification { def up = Mock(EthereumMultistream) { 1 * getHead(Selector.empty) >> head } - ConnectNewHeads connectNewHeads = new ConnectNewHeads(up, Schedulers.boundedElastic()) + ConnectNewHeads connectNewHeads = new ConnectNewHeads(up, Schedulers.parallel()) when: def act1 = connectNewHeads.connect(Selector.empty) def act2 = connectNewHeads.connect(Selector.empty) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index a1d00114..3798883d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -90,7 +90,7 @@ class EthereumGrpcUpstreamSpec extends Specification { ) } }) - def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic()) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel()) upstream.setLag(0) upstream.update( BlockchainOuterClass.DescribeChain.newBuilder() @@ -161,7 +161,7 @@ class EthereumGrpcUpstreamSpec extends Specification { }).start() } }) - def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic()) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel()) upstream.setLag(0) upstream.update( BlockchainOuterClass.DescribeChain.newBuilder() @@ -233,7 +233,7 @@ class EthereumGrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.boundedElastic()) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel()) upstream.setLag(0) upstream.update( BlockchainOuterClass.DescribeChain.newBuilder() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index 99b3a128..6e518ac6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -70,7 +70,7 @@ class GrpcHeadSpec extends Specification { convert, null, new MostWorkForkChoice(), - Schedulers.boundedElastic() + Schedulers.parallel() ) when: def act = head.getFlux() @@ -137,7 +137,7 @@ class GrpcHeadSpec extends Specification { convert, null, new MostWorkForkChoice(), - Schedulers.boundedElastic() + Schedulers.parallel() ) when: def act = head.getFlux()