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

Fix Flux.merge() concurrency limit in SubscribeChainStatus (#792)

Browse Source
This commit is contained in:
Anton
2026-02-06 15:41:04 +03:00
committed by GitHub
parent b375fd34b4
commit 15047c7d26
1 changed files with 19 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeChainStatus.kt
View File

@@ -25,25 +25,25 @@ class SubscribeChainStatus(
} }
fun chainStatuses(): Flux<BlockchainOuterClass.SubscribeChainStatusResponse> { fun chainStatuses(): Flux<BlockchainOuterClass.SubscribeChainStatusResponse> {
return Flux.merge( // we need to track not only multistreams with upstreams but all of them
// we need to track not only multistreams with upstreams but all of them // because upstreams can be added in runtime with hot config reload
// because upstreams can be added in runtime with hot config reload val sources = multistreamHolder.all()
multistreamHolder.all() .filter { Common.ChainRef.forNumber(it.getChain().id) != null }
.filter { Common.ChainRef.forNumber(it.getChain().id) != null } .map { ms ->
.map { ms -> Flux.concat(
Flux.concat( // the first event must be filled with all fields
// the first event must be filled with all fields firstFullEvent(ms),
firstFullEvent(ms), Flux.merge(
Flux.merge( // head events are separated from others
// head events are separated from others headEvents(ms),
headEvents(ms), multistreamEvents(ms),
multistreamEvents(ms), ),
), )
) }
}, return Flux.merge(Flux.fromIterable(sources), sources.size)
).doOnError { .doOnError {
log.error("Error during sending chain statuses", it) log.error("Error during sending chain statuses", it)
} }
} }
private fun multistreamEvents(ms: Multistream): Flux<BlockchainOuterClass.SubscribeChainStatusResponse> { private fun multistreamEvents(ms: Multistream): Flux<BlockchainOuterClass.SubscribeChainStatusResponse> {