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

problem: interval throws error if it's not consumed fast enough

Browse Source
This commit is contained in:
Igor Artamonov
2019-09-05 00:29:41 -04:00
parent d553560be2
commit fbc6e5588b
1 changed files with 8 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -125,10 +125,10 @@ class NativeCall(
val all = ctx.getApis().toFlux().share() val all = ctx.getApis().toFlux().share()
//execute on the first API immediately, and then make a delay between each call to not overload upstreams //execute on the first API immediately, and then make a delay between each call to not overload upstreams
val immediate = Flux.from(all).take(1) val immediate = Flux.from(all).take(1)
val repeatControl = EmitterProcessor.create<Boolean>()//TopicProcessor.create<Boolean>() val repeatControl = EmitterProcessor.create<Boolean>()
val retries = Flux.from(all).skip(1) val retries = Flux.from(all).skip(1)
.zipWith(repeatControl).zipWith(Flux.interval(Duration.ofMillis(200))) .zipWith(repeatControl.delayElements(Duration.ofMillis(200))) //manages when need another call, make delay for at least of 200ms between calls
.map { it.t1.t1 } .map { it.t1 }
return Flux.concat(immediate, retries) return Flux.concat(immediate, retries)
.flatMap { api -> .flatMap { api ->
@@ -143,6 +143,11 @@ class NativeCall(
} }
res res
}) })
.doOnNext {
if (!it.isResolved()) {
log.debug("No quorum for ${ctx.payload.method} as ${ctx.callQuorum}")
}
}
.filter { it.isResolved() } .filter { it.isResolved() }
.map { .map {
val result = it.getResult() val result = it.getResult()