From fbc6e5588b3eea12aa619c70a5a9b3d955f92a18 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 5 Sep 2019 00:29:41 -0400 Subject: [PATCH] problem: interval throws error if it's not consumed fast enough --- .../kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index bfb9707f..b65e79d0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -125,10 +125,10 @@ class NativeCall( val all = ctx.getApis().toFlux().share() //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 repeatControl = EmitterProcessor.create()//TopicProcessor.create() + val repeatControl = EmitterProcessor.create() val retries = Flux.from(all).skip(1) - .zipWith(repeatControl).zipWith(Flux.interval(Duration.ofMillis(200))) - .map { it.t1.t1 } + .zipWith(repeatControl.delayElements(Duration.ofMillis(200))) //manages when need another call, make delay for at least of 200ms between calls + .map { it.t1 } return Flux.concat(immediate, retries) .flatMap { api -> @@ -143,6 +143,11 @@ class NativeCall( } res }) + .doOnNext { + if (!it.isResolved()) { + log.debug("No quorum for ${ctx.payload.method} as ${ctx.callQuorum}") + } + } .filter { it.isResolved() } .map { val result = it.getResult()