From b97108fd35bf1eef8a18b907b4d6aab761e1ca63 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 21 Apr 2023 18:02:25 +0800 Subject: [PATCH] use head scheduler instead of stock schedulers in other parts of system --- .../dshackle/startup/ConfiguredUpstreams.kt | 1 + .../ethereum/EthereumEgressSubscription.kt | 6 ++++-- .../upstream/ethereum/EthereumMultistream.kt | 4 ++-- .../ethereum/EthereumWsConnectionFactory.kt | 4 +++- .../upstream/ethereum/EthereumWsHead.kt | 5 ++--- .../upstream/ethereum/WsConnectionImpl.kt | 9 +++++---- .../ethereum/subscribe/ConnectBlockUpdates.kt | 9 ++++----- .../upstream/ethereum/subscribe/ConnectLogs.kt | 7 ++----- .../ethereum/subscribe/ConnectNewHeads.kt | 12 ++++-------- .../ethereum/subscribe/ConnectSyncing.kt | 5 ----- .../ethereum_pos/EthereumPosMultiStream.kt | 4 ++-- .../EthereumEgressSubscriptionSpec.groovy | 17 +++++++++-------- .../ethereum/WsConnectionImplRealSpec.groovy | 7 +++++-- .../ethereum/WsConnectionImplSpec.groovy | 10 +++++++--- .../subscribe/ConnectBlockUpdatesSpec.groovy | 11 ++++++----- .../ethereum/subscribe/ConnectLogsSpec.groovy | 9 +++++---- .../subscribe/ConnectNewHeadsSpec.groovy | 3 ++- 17 files changed, 63 insertions(+), 60 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index c46d1fb5..20c96e8f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -388,6 +388,7 @@ open class ConfiguredUpstreams( id, chain, endpoint.url, endpoint.origin ?: URI("http://localhost"), + headScheduler ).apply { config = endpoint basicAuth = endpoint.basicAuth diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt index 41fe519c..ba9c7d6b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt @@ -10,9 +10,11 @@ import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.hex.Hex32 import org.slf4j.LoggerFactory import reactor.core.publisher.Flux +import reactor.core.scheduler.Scheduler open class EthereumEgressSubscription( val upstream: EthereumLikeMultistream, + val scheduler: Scheduler, val pendingTxesSource: PendingTxesSource? ) : EgressSubscription { @@ -37,8 +39,8 @@ open class EthereumEgressSubscription( } } - private val newHeads = ConnectNewHeads(upstream) - open val logs = ConnectLogs(upstream) + private val newHeads = ConnectNewHeads(upstream, scheduler) + open val logs = ConnectLogs(upstream, scheduler) private val syncing = ConnectSyncing(upstream) override fun getAvailableTopics() = availableTopics diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 19103a9e..7d035f15 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -63,7 +63,7 @@ open class EthereumMultistream( ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK) private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory(), tracer) - private var subscribe = EthereumEgressSubscription(this, NoPendingTxes()) + private var subscribe = EthereumEgressSubscription(this, headScheduler, NoPendingTxes()) private val supportsEIP1559 = when (chain) { Chain.ETHEREUM, Chain.TESTNET_ROPSTEN, @@ -102,7 +102,7 @@ open class EthereumMultistream( AggregatedPendingTxes(it) } } - subscribe = EthereumEgressSubscription(this, pendingTxes) + subscribe = EthereumEgressSubscription(this, headScheduler, pendingTxes) } override fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt index a2228df3..0d3eb5f6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt @@ -8,6 +8,7 @@ import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Tag import io.micrometer.core.instrument.Timer +import reactor.core.scheduler.Scheduler import java.net.URI open class EthereumWsConnectionFactory( @@ -15,6 +16,7 @@ open class EthereumWsConnectionFactory( private val chain: Chain, private val uri: URI, private val origin: URI, + private val scheduler: Scheduler ) { var basicAuth: AuthConfig.ClientBasicAuth? = null @@ -42,7 +44,7 @@ open class EthereumWsConnectionFactory( } open fun createWsConnection(connIndex: Int = 0, onDisconnect: () -> Unit): WsConnection = - WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), onDisconnect).also { ws -> + WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), onDisconnect, scheduler).also { ws -> config?.frameSize?.let { ws.frameSize = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index d791e40b..216d86d3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -33,7 +33,6 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.Sinks import reactor.core.scheduler.Scheduler -import reactor.core.scheduler.Schedulers import reactor.retry.Repeat import java.time.Duration @@ -45,7 +44,7 @@ class EthereumWsHead( private val wsSubscriptions: WsSubscriptions, private val skipEnhance: Boolean, private val wsConnectionResubscribeScheduler: Scheduler, - headScheduler: Scheduler, + private val headScheduler: Scheduler, ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator, headScheduler), Lifecycle { private var connectionId: String? = null @@ -119,7 +118,7 @@ class EthereumWsHead( } .flatMap(JsonRpcResponse::requireResult) .map { BlockContainer.fromEthereumJson(it, upstreamId) } - .subscribeOn(Schedulers.boundedElastic()) + .subscribeOn(headScheduler) .timeout(Defaults.timeoutInternal, Mono.empty()) }.repeatWhenEmpty { n -> Repeat.times(5) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt index e18f31f6..2aab647f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt @@ -44,7 +44,7 @@ import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.Sinks -import reactor.core.scheduler.Schedulers +import reactor.core.scheduler.Scheduler import reactor.netty.http.client.HttpClient import reactor.netty.http.client.WebsocketClientSpec import reactor.netty.http.websocket.WebsocketInbound @@ -67,7 +67,8 @@ open class WsConnectionImpl( private val origin: URI, private val basicAuth: AuthConfig.ClientBasicAuth?, private val rpcMetrics: RpcMetrics?, - private val onDisconnect: () -> Unit + private val onDisconnect: () -> Unit, + private val scheduler: Scheduler ) : AutoCloseable, WsConnection, Cloneable { companion object { @@ -292,8 +293,8 @@ open class WsConnectionImpl( return outbound.send( Flux.merge( - calls.subscribeOn(Schedulers.boundedElastic()), - consumer.then(Mono.empty()).subscribeOn(Schedulers.boundedElastic()) + calls.subscribeOn(scheduler), + consumer.then(Mono.empty()).subscribeOn(scheduler) ) ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt index 0a3f2899..fd9adec1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt @@ -22,9 +22,8 @@ import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.SubscriptionConnect import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream -import org.slf4j.LoggerFactory import reactor.core.publisher.Flux -import reactor.core.scheduler.Schedulers +import reactor.core.scheduler.Scheduler import java.time.Duration import java.util.LinkedList import java.util.concurrent.ConcurrentHashMap @@ -33,11 +32,11 @@ import kotlin.concurrent.read import kotlin.concurrent.write class ConnectBlockUpdates( - private val upstream: EthereumLikeMultistream + private val upstream: EthereumLikeMultistream, + private val scheduler: Scheduler ) : SubscriptionConnect { companion object { - private val log = LoggerFactory.getLogger(ConnectBlockUpdates::class.java) private const val HISTORY_LIMIT = 6 * 3 } @@ -53,7 +52,7 @@ class ConnectBlockUpdates( override fun connect(matcher: Selector.Matcher): Flux { return connected.computeIfAbsent(matcher.describeInternal()) { key -> extract(upstream.getHead(matcher)) - .publishOn(Schedulers.boundedElastic()) + .publishOn(scheduler) .publish() .refCount(1, Duration.ofSeconds(60)) .doFinally { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt index 071b4301..044d9d3d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt @@ -22,8 +22,8 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.hex.Hex32 import io.emeraldpay.etherjar.hex.HexDataComparator -import org.slf4j.LoggerFactory import reactor.core.publisher.Flux +import reactor.core.scheduler.Scheduler import java.util.function.Function open class ConnectLogs( @@ -32,14 +32,11 @@ open class ConnectLogs( ) { companion object { - private val log = LoggerFactory.getLogger(ConnectLogs::class.java) - private val ADDR_COMPARATOR = HexDataComparator() private val TOPIC_COMPARATOR = HexDataComparator() } - constructor(upstream: EthereumLikeMultistream) : this(upstream, ConnectBlockUpdates(upstream)) - + constructor(upstream: EthereumLikeMultistream, scheduler: Scheduler) : this(upstream, ConnectBlockUpdates(upstream, scheduler)) private val produceLogs = ProduceLogs(upstream) fun start(matcher: Selector.Matcher): Flux { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt index d49e4f75..4bd747de 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt @@ -19,9 +19,8 @@ import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.SubscriptionConnect import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage -import org.slf4j.LoggerFactory import reactor.core.publisher.Flux -import reactor.core.scheduler.Schedulers +import reactor.core.scheduler.Scheduler import java.time.Duration import java.util.concurrent.ConcurrentHashMap @@ -29,20 +28,17 @@ import java.util.concurrent.ConcurrentHashMap * Connects/reconnects to the upstream to produce NewHeads messages */ class ConnectNewHeads( - private val upstream: EthereumLikeMultistream + private val upstream: EthereumLikeMultistream, + private val scheduler: Scheduler ) : SubscriptionConnect { - companion object { - private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java) - } - private val connected: MutableMap> = ConcurrentHashMap() override fun connect(matcher: Selector.Matcher): Flux = connected.computeIfAbsent(matcher.describeInternal()) { key -> ProduceNewHeads(upstream.getHead(matcher)) .start() - .publishOn(Schedulers.boundedElastic()) + .publishOn(scheduler) .publish() .refCount(1, Duration.ofSeconds(60)) .doFinally { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt index 41aeac4d..1c53c96b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt @@ -19,7 +19,6 @@ import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.SubscriptionConnect import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream -import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import java.time.Duration import java.util.concurrent.locks.ReentrantLock @@ -29,10 +28,6 @@ class ConnectSyncing( private val upstream: EthereumLikeMultistream ) : SubscriptionConnect { - companion object { - private val log = LoggerFactory.getLogger(ConnectSyncing::class.java) - } - private var connected: Flux? = null private val connectLock = ReentrantLock() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index 75f0e3f6..f90c570a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -58,7 +58,7 @@ open class EthereumPosMultiStream( ) private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory(), tracer) - private var subscribe = EthereumEgressSubscription(this, NoPendingTxes()) + private var subscribe = EthereumEgressSubscription(this, headScheduler, NoPendingTxes()) private val feeEstimation = EthereumPriorityFees(this, reader, 256) private val filteredHeads: MutableMap = ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK) @@ -198,6 +198,6 @@ open class EthereumPosMultiStream( AggregatedPendingTxes(it) } } - subscribe = EthereumEgressSubscription(this, pendingTxes) + subscribe = EthereumEgressSubscription(this, headScheduler, pendingTxes) } } 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 efc02a7d..0a384473 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy @@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.hex.Hex32 +import reactor.core.scheduler.Schedulers import spock.lang.Specification class EthereumEgressSubscriptionSpec extends Specification { def "read empty logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([:]) @@ -36,7 +37,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read single address logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: "0x829bd824b016326a401d083b33d092293333a830" @@ -61,7 +62,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "ignores invalid address for logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: "829bd824b016326a401d083b33d092293333a830" @@ -74,7 +75,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read multi address logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: ["0x829bd824b016326a401d083b33d092293333a830", "0x401d083b33d092293333a83829bd824b016326a0"] @@ -90,7 +91,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read single topic logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ topics: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" @@ -115,7 +116,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read invalid topic for request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ topics: [ @@ -133,7 +134,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read multi topic logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ topics: [ @@ -152,7 +153,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def "read full logs request"() { setup: - def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource)) + def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) when: def act = ethereumSubscribe.readLogsRequest([ address: "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695", diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy index 7eceb26b..c15646ac 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy @@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.test.MockWSServer import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Shared import spock.lang.Specification @@ -38,7 +39,8 @@ class WsConnectionImplRealSpec extends Specification { "test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), - "http://localhost:${port}".toURI() + "http://localhost:${port}".toURI(), + Schedulers.parallel() ) ).create(null).getConnection() } @@ -117,7 +119,8 @@ class WsConnectionImplRealSpec extends Specification { "test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), - "http://localhost:${port}".toURI() + "http://localhost:${port}".toURI(), + Schedulers.parallel() ) ).create(up).getConnection() when: 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 c8a56dfc..ce2eb9c8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy @@ -23,6 +23,7 @@ import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.TransactionJson import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -39,7 +40,8 @@ class WsConnectionImplSpec extends Specification { "test", Chain.ETHEREUM, new URI("http://localhost"), - new URI("http://localhost") + new URI("http://localhost"), + Schedulers.parallel() ) ) def apiMock = TestingCommons.api() @@ -73,7 +75,8 @@ class WsConnectionImplSpec extends Specification { "test", Chain.ETHEREUM, new URI("http://localhost"), - new URI("http://localhost") + new URI("http://localhost"), + Schedulers.parallel() ) ) def apiMock = TestingCommons.api() @@ -105,7 +108,8 @@ class WsConnectionImplSpec extends Specification { "test", Chain.ETHEREUM, new URI("http://localhost"), - new URI("http://localhost") + new URI("http://localhost"), + Schedulers.parallel() ) ) def apiMock = TestingCommons.api() 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 2307e774..fa343cf2 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 @@ -26,6 +26,7 @@ import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import spock.lang.Specification import java.time.Duration @@ -37,7 +38,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Extracts updates"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) def block = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -71,7 +72,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Produce DROP updates for replaced block"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) def block = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -105,7 +106,7 @@ class ConnectBlockUpdatesSpec extends Specification { def "Gets prev version if available"() { setup: - def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) def block1 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -169,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)) + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream), Schedulers.boundedElastic()) def block1 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") number = 13412871 @@ -234,7 +235,7 @@ class ConnectBlockUpdatesSpec extends Specification { def up = Mock(EthereumMultistream) { 1 * getHead(Selector.empty) >> head } - def connectBlockUpdates = new ConnectBlockUpdates(up) + def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.boundedElastic()) 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 9e2034b5..0de35e85 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 @@ -24,6 +24,7 @@ import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.hex.Hex32 import io.emeraldpay.etherjar.hex.HexData import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import spock.lang.Specification class ConnectLogsSpec extends Specification { @@ -90,7 +91,7 @@ class ConnectLogsSpec extends Specification { def "Filter is empty"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) when: def input = Flux.fromIterable([ log1, log2, log3, log4 @@ -108,7 +109,7 @@ class ConnectLogsSpec extends Specification { def "Filter by address"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) when: def input = Flux.fromIterable([ log1, log2 @@ -123,7 +124,7 @@ class ConnectLogsSpec extends Specification { def "Filter by topic"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) when: def input = Flux.fromIterable([ log1, log2, log3, log4 @@ -140,7 +141,7 @@ class ConnectLogsSpec extends Specification { def "Filter by address and topic"() { setup: - def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream) + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Schedulers.boundedElastic()) 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 25781ccc..998640c3 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 @@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -20,7 +21,7 @@ class ConnectNewHeadsSpec extends Specification { def up = Mock(EthereumMultistream) { 1 * getHead(Selector.empty) >> head } - ConnectNewHeads connectNewHeads = new ConnectNewHeads(up) + ConnectNewHeads connectNewHeads = new ConnectNewHeads(up, Schedulers.boundedElastic()) when: def act1 = connectNewHeads.connect(Selector.empty) def act2 = connectNewHeads.connect(Selector.empty)