add schedulers config

src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt

@@ -16,7 +16,6 @@
  */
 package io.emeraldpay.dshackle
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder
 import io.emeraldpay.dshackle.config.MainConfig
 import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerGrpc
 import io.grpc.Server
@@ -24,7 +23,7 @@ import io.grpc.netty.NettyServerBuilder
 import io.micrometer.core.instrument.Metrics
 import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics
 import org.slf4j.LoggerFactory
-import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import org.springframework.stereotype.Service
 import java.net.InetSocketAddress
 import java.util.concurrent.Executors
@@ -33,10 +32,10 @@ import javax.annotation.PreDestroy
 
 @Service
 open class GrpcServer(
-    @Autowired val rpcs: List<io.grpc.BindableService>,
+    private val rpcs: List<io.grpc.BindableService>,
-    @Autowired val mainConfig: MainConfig,
+    private val mainConfig: MainConfig,
-    @Autowired val tlsSetup: TlsSetup,
+    private val tlsSetup: TlsSetup,
-    @Autowired val accessHandler: AccessHandlerGrpc
+    private val accessHandler: AccessHandlerGrpc
 ) {
 
     private val log = LoggerFactory.getLogger(GrpcServer::class.java)
@@ -67,7 +66,7 @@ open class GrpcServer(
             serverBuilder.addService(it)
         }
 
-        val pool = Executors.newFixedThreadPool(20, ThreadFactoryBuilder().setNameFormat("fixed-grpc-%d").build())
+        val pool = Executors.newFixedThreadPool(20, CustomizableThreadFactory("fixed-grpc-%d"))
 
         serverBuilder.executor(
             if (mainConfig.monitoring.enableExtended)

src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt

@@ -0,0 +1,40 @@
+package io.emeraldpay.dshackle.config.context
+
+import io.emeraldpay.dshackle.config.MonitoringConfig
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory
+import reactor.core.scheduler.Scheduler
+import reactor.core.scheduler.Schedulers
+import java.util.concurrent.Executors
+
+@Configuration
+open class SchedulersConfig {
+    @Bean
+    open fun rpcScheduler(monitoringConfig: MonitoringConfig): Scheduler {
+        return makeScheduler("blockchain-rpc-scheduler", "blockchain_rpc", 30, monitoringConfig)
+    }
+
+    @Bean
+    open fun trackTxScheduler(monitoringConfig: MonitoringConfig): Scheduler {
+        return makeScheduler("tracktx-scheduler", "tracktx", 5, monitoringConfig)
+    }
+
+    private fun makeScheduler(name: String, prefix: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler {
+        val pool = Executors.newFixedThreadPool(size, CustomizableThreadFactory("$name-%d"))
+
+        return Schedulers.fromExecutorService(
+            if (monitoringConfig.enableExtended)
+                ExecutorServiceMetrics.monitor(
+                    Metrics.globalRegistry,
+                    pool,
+                    name,
+                    prefix
+                )
+            else
+                pool
+        )
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt

@@ -26,11 +26,12 @@ import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Metrics
 import io.micrometer.core.instrument.Timer
 import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.context.annotation.DependsOn
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.core.scheduler.Schedulers
+import reactor.core.scheduler.Scheduler
 import java.util.Locale
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
@@ -46,6 +47,8 @@ class BlockchainRpc(
     private val describe: Describe,
     private val subscribeStatus: SubscribeStatus,
     private val estimateFee: EstimateFee,
+    @Qualifier("rpcScheduler")
+    private val scheduler: Scheduler
 ) : ReactorBlockchainGrpc.BlockchainImplBase() {
 
     private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)
@@ -69,7 +72,7 @@ class BlockchainRpc(
         val idsMap = mutableMapOf<Int, String>()
         return nativeCall.nativeCall(
             request
-                .subscribeOn(Schedulers.boundedElastic())
+                .subscribeOn(scheduler)
                 .doOnNext { req ->
                     metrics = chainMetrics.get(req.chain)
                     metrics?.let { m ->
@@ -115,7 +118,7 @@ class BlockchainRpc(
     }
 
     override fun subscribeTxStatus(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
-        return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request ->
+        return requestMono.subscribeOn(scheduler).flatMapMany { request ->
             val chain = Chain.byId(request.chainValue)
             val metrics = chainMetrics.get(chain)
             metrics.subscribeTxMetric.increment()
@@ -128,13 +131,13 @@ class BlockchainRpc(
             } catch (t: Throwable) {
                 log.error("Internal error during Tx Subscription", t)
                 failMetric.increment()
-                Flux.error<BlockchainOuterClass.TxStatus>(IllegalStateException("Internal Error"))
+                Flux.error(IllegalStateException("Internal Error"))
             }
         }
     }
 
     override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
-        return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request ->
+        return requestMono.subscribeOn(scheduler).flatMapMany { request ->
             val chain = Chain.byId(request.asset.chainValue)
             val metrics = chainMetrics.get(chain)
             metrics.subscribeBalanceMetric.increment()
@@ -151,13 +154,13 @@ class BlockchainRpc(
             } catch (t: Throwable) {
                 log.error("Internal error during Balance Subscription", t)
                 failMetric.increment()
-                Flux.error<BlockchainOuterClass.AddressBalance>(IllegalStateException("Internal Error"))
+                Flux.error(IllegalStateException("Internal Error"))
             }
         }
     }
 
     override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
-        return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request ->
+        return requestMono.subscribeOn(scheduler).flatMapMany { request ->
             val chain = Chain.byId(request.asset.chainValue)
             val metrics = chainMetrics.get(chain)
             metrics.getBalanceMetric.increment()
@@ -186,7 +189,7 @@ class BlockchainRpc(
 
     override fun estimateFee(request: Mono<BlockchainOuterClass.EstimateFeeRequest>): Mono<BlockchainOuterClass.EstimateFeeResponse> {
         return request
-            .subscribeOn(Schedulers.boundedElastic())
+            .subscribeOn(scheduler)
             .flatMap {
                 val chain = Chain.byId(it.chainValue)
                 val metrics = chainMetrics.get(chain)

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt

@@ -32,12 +32,11 @@ import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
-import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Scheduler
-import reactor.core.scheduler.Schedulers
 import reactor.util.retry.Retry
 import java.math.BigInteger
 import java.time.Duration
@@ -47,7 +46,9 @@ import kotlin.math.min
 
 @Service
 class TrackEthereumTx(
-    @Autowired private val multistreamHolder: MultistreamHolder
+    private val multistreamHolder: MultistreamHolder,
+    @Qualifier("trackTxScheduler")
+    private val scheduler: Scheduler
 ) : TrackTx {
 
     companion object {
@@ -57,8 +58,6 @@ class TrackEthereumTx(
         private val NOT_MINED_TRACK_TTL = NOT_FOUND_TRACK_TTL.multipliedBy(2)
     }
 
-    var scheduler: Scheduler = Schedulers.boundedElastic()
-
     private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java)
 
     override fun isSupported(chain: Chain): Boolean {

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy

@@ -35,6 +35,7 @@ import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import reactor.core.publisher.Flux
+import reactor.core.scheduler.Schedulers
 import reactor.test.StepVerifier
 import reactor.test.scheduler.VirtualTimeScheduler
 import spock.lang.Specification
@@ -99,7 +100,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)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         apiMock.answer("eth_getTransactionByHash", [txId], txJson)
         apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson)
@@ -122,9 +123,8 @@ class TrackEthereumTxSpec extends Specification {
         ((EthereumPosMultiStream) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) {
             _ * getFlux() >> Flux.empty()
         }
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
         def scheduler = VirtualTimeScheduler.create(true)
-        trackTx.scheduler = scheduler
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, scheduler)
 
         apiMock.answer("eth_getTransactionByHash", [txId], null)
 
@@ -169,9 +169,8 @@ 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)
         def scheduler = VirtualTimeScheduler.create(true)
-        trackTx.scheduler = scheduler
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, scheduler)
 
         apiMock.answerOnce("eth_getTransactionByHash", [txId], null)
         apiMock.answer("eth_getTransactionByHash", [txId], txJson)
@@ -194,7 +193,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)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6)
         def block = new BlockContainer(
@@ -216,7 +215,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)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6)
         def block = new BlockContainer(
@@ -289,7 +288,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)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
 
         apiMock.answerOnce("eth_getTransactionByHash", [txId], null)
         apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted)