diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 842ffabb..d409ea1e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -17,6 +17,7 @@ import reactor.core.publisher.toMono import reactor.util.function.Tuple2 import reactor.util.function.Tuples import java.lang.Exception +import java.time.Duration import java.util.function.Predicate @Service @@ -30,7 +31,9 @@ class NativeCall( open fun nativeCall(requestMono: Mono): Flux { return requestMono.flatMapMany(this::prepareCall) .map(this::setupCallParams) + .parallel() .flatMap(this::executeOnRemote) + .sequential() .map(this::buildResponse) .doOnError { e -> log.warn("Error during native call", e) } .onErrorResume(this::processException) @@ -89,7 +92,11 @@ class NativeCall( fun executeOnRemote(ctx: CallContext>>): Mono> { val p: Predicate = CallQuorum.untilResolved(ctx.callQuorum) - return ctx.apis.toFlux() + val all = ctx.apis.toFlux().share() + //execute on the first API immediately, and then make a delay between each call to not dos upstreams + val immediate = Flux.from(all).take(1) + val retries = Flux.from(all).delayElements(Duration.ofMillis(200)) + return Flux.concat(immediate, retries) .takeWhile(p) .flatMap { api -> api.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { Tuples.of(it, api.upstream!!) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index f9cdba8b..9d0c8a7d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -93,7 +93,7 @@ class EthereumGrpcTransportSpec extends Specification { then: 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr - 1 * otherSideAggr.getApis(_) >> [otherSideApi].multiply(3).iterator() + 1 * otherSideAggr.getApis(_) >> [otherSideApi].multiply(34).iterator() _ * otherSideAggr.getHead() >> Stub(EthereumHead) _ * otherSideAggr.getTargets() >> ethereumTargets status.failed == 0