use head scheduler instead of stock schedulers in other parts of system

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

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)
+    private val newHeads = ConnectNewHeads(upstream, scheduler)
-    open val logs = ConnectLogs(upstream)
+    open val logs = ConnectLogs(upstream, scheduler)
     private val syncing = ConnectSyncing(upstream)
 
     override fun getAvailableTopics() = availableTopics

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

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
             }

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<Any>(5)

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()),
+                calls.subscribeOn(scheduler),
-                consumer.then(Mono.empty<ByteBuf>()).subscribeOn(Schedulers.boundedElastic())
+                consumer.then(Mono.empty<ByteBuf>()).subscribeOn(scheduler)
             )
         )
     }

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<ConnectBlockUpdates.Update> {
 
     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<Update> {
         return connected.computeIfAbsent(matcher.describeInternal()) { key ->
             extract(upstream.getHead(matcher))
-                .publishOn(Schedulers.boundedElastic())
+                .publishOn(scheduler)
                 .publish()
                 .refCount(1, Duration.ofSeconds(60))
                 .doFinally {

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<LogMessage> {

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<NewHeadMessage> {
 
-    companion object {
-        private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java)
-    }
-
     private val connected: MutableMap<String, Flux<NewHeadMessage>> = ConcurrentHashMap()
 
     override fun connect(matcher: Selector.Matcher): Flux<NewHeadMessage> =
         connected.computeIfAbsent(matcher.describeInternal()) { key ->
             ProduceNewHeads(upstream.getHead(matcher))
                 .start()
-                .publishOn(Schedulers.boundedElastic())
+                .publishOn(scheduler)
                 .publish()
                 .refCount(1, Duration.ofSeconds(60))
                 .doFinally {

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<Boolean> {
 
-    companion object {
-        private val log = LoggerFactory.getLogger(ConnectSyncing::class.java)
-    }
-
     private var connected: Flux<Boolean>? = null
     private val connectLock = ReentrantLock()
 

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<String, Head> =
         ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
@@ -198,6 +198,6 @@ open class EthereumPosMultiStream(
                     AggregatedPendingTxes(it)
                 }
             }
-        subscribe = EthereumEgressSubscription(this, pendingTxes)
+        subscribe = EthereumEgressSubscription(this, headScheduler, pendingTxes)
     }
 }

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

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:

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

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

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

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)