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

use parallel scheduler in tests for validate blocking operations

Browse Source
This commit is contained in:
a10zn8
2023-04-21 18:06:24 +08:00
parent b97108fd35
commit dca0e6614f
18 changed files with 59 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy
View File

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

20
src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy
View File

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

4
src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
View File

@@ -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<EthereumPosRpcUpstream> 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<EthereumPosRpcUpstream> upstreams) {

6
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

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

2
src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy
View File

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

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

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

2
src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy
View File

@@ -117,7 +117,7 @@ class HeadLagObserverSpec extends Specification {
TestHeadLagObserver(@NotNull Head master, @NotNull Collection<? extends Upstream> followers) {
super(master, followers, DistanceExtractor.@Companion::extractPowDistance,
Schedulers.boundedElastic(), Duration.ofNanos(1))
Schedulers.parallel(), Duration.ofNanos(1))
}
@Override

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

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

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

@@ -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<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches, Schedulers.boundedElastic(), TestingCommons.tracerMock())
super(chain, upstreams, caches, Schedulers.parallel(), TestingCommons.tracerMock())
}
@NotNull

2
src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHeadSpec.groovy
View File

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

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy
View File

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

16
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
View File

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

10
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy
View File

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

10
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy
View File

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

8
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy
View File

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

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy
View File

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

6
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy
View File

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

4
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy
View File

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