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

Merge pull request #58 from p2p-org/use-another-scheduler

Browse Source 
try to use another scheduler for request processing
This commit is contained in:
Vyacheslav Shebanov
2022-12-05 12:43:30 +02:00
committed by GitHub
parent d50684f71b f93bccf9f5
commit bfc4cc620e
1 changed files with 14 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -26,11 +26,11 @@ import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Timer import io.micrometer.core.instrument.Timer
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.context.annotation.DependsOn import org.springframework.context.annotation.DependsOn
import org.springframework.stereotype.Service import org.springframework.stereotype.Service
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers
import java.util.Locale import java.util.Locale
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
@@ -38,14 +38,14 @@ import java.util.concurrent.TimeUnit
@Service @Service
@DependsOn("monitoringSetup") @DependsOn("monitoringSetup")
class BlockchainRpc( class BlockchainRpc(
@Autowired private val nativeCall: NativeCall, private val nativeCall: NativeCall,
@Autowired private val nativeSubscribe: NativeSubscribe, private val nativeSubscribe: NativeSubscribe,
@Autowired private val streamHead: StreamHead, private val streamHead: StreamHead,
@Autowired private val trackTx: List<TrackTx>, private val trackTx: List<TrackTx>,
@Autowired private val trackAddress: List<TrackAddress>, private val trackAddress: List<TrackAddress>,
@Autowired private val describe: Describe, private val describe: Describe,
@Autowired private val subscribeStatus: SubscribeStatus, private val subscribeStatus: SubscribeStatus,
@Autowired private val estimateFee: EstimateFee, private val estimateFee: EstimateFee,
) : ReactorBlockchainGrpc.BlockchainImplBase() { ) : ReactorBlockchainGrpc.BlockchainImplBase() {
private val log = LoggerFactory.getLogger(BlockchainRpc::class.java) private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)
@@ -69,6 +69,7 @@ class BlockchainRpc(
val idsMap = mutableMapOf<Int, String>() val idsMap = mutableMapOf<Int, String>()
return nativeCall.nativeCall( return nativeCall.nativeCall(
request request
.subscribeOn(Schedulers.boundedElastic())
.doOnNext { req -> .doOnNext { req ->
metrics = chainMetrics.get(req.chain) metrics = chainMetrics.get(req.chain)
metrics?.let { m -> metrics?.let { m ->
@@ -114,7 +115,7 @@ class BlockchainRpc(
} }
override fun subscribeTxStatus(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> { override fun subscribeTxStatus(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
return requestMono.flatMapMany { request -> return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request ->
val chain = Chain.byId(request.chainValue) val chain = Chain.byId(request.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)
metrics.subscribeTxMetric.increment() metrics.subscribeTxMetric.increment()
@@ -133,7 +134,7 @@ class BlockchainRpc(
} }
override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> { override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
return requestMono.flatMapMany { request -> return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request ->
val chain = Chain.byId(request.asset.chainValue) val chain = Chain.byId(request.asset.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)
metrics.subscribeBalanceMetric.increment() metrics.subscribeBalanceMetric.increment()
@@ -156,7 +157,7 @@ class BlockchainRpc(
} }
override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> { override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
return requestMono.flatMapMany { request -> return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request ->
val chain = Chain.byId(request.asset.chainValue) val chain = Chain.byId(request.asset.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)
metrics.getBalanceMetric.increment() metrics.getBalanceMetric.increment()
@@ -185,6 +186,7 @@ class BlockchainRpc(
override fun estimateFee(request: Mono<BlockchainOuterClass.EstimateFeeRequest>): Mono<BlockchainOuterClass.EstimateFeeResponse> { override fun estimateFee(request: Mono<BlockchainOuterClass.EstimateFeeRequest>): Mono<BlockchainOuterClass.EstimateFeeResponse> {
return request return request
.subscribeOn(Schedulers.boundedElastic())
.flatMap { .flatMap {
val chain = Chain.byId(it.chainValue) val chain = Chain.byId(it.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)