From 913ad6965a9055f9b215b890d97d9bbf0aedfb4e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 30 Jul 2019 00:21:31 -0400 Subject: [PATCH] problem: can notify client about new head before actual upstream made available --- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 23 +++++++++++++++---- .../dshackle/upstream/ChainUpstreams.kt | 3 +++ .../dshackle/upstream/EthereumRpcHead.kt | 4 +--- .../dshackle/upstream/EthereumWs.kt | 3 +-- .../dshackle/upstream/GrpcUpstream.kt | 2 +- .../dshackle/upstream/UpstreamServices.kt | 18 +++++++++++++++ 6 files changed, 42 insertions(+), 11 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 3b30a299..85d54058 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -4,6 +4,8 @@ import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.upstream.AvailableChains +import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.UpstreamServices import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.TransactionId @@ -16,6 +18,7 @@ import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import java.lang.Exception +import java.time.Duration import java.util.concurrent.ConcurrentLinkedQueue import javax.annotation.PostConstruct import kotlin.collections.HashMap @@ -56,10 +59,16 @@ class StreamHead( private fun onBlock(chain: Chain, block: BlockJson) { log.info("New block ${block.number} on ${chain.chainCode}") - clients[chain]!!.toFlux() - .subscribe { stream -> - notify(chain, block, stream) + upstreams.getUpstream(chain)?.let { up -> + UpstreamServices.onceOk(up).subscribe {avail -> + if (avail) { + clients[chain]!!.toFlux() + .subscribe { stream -> + notify(chain, block, stream) + } } + } + } } fun add(requestMono: Mono): Flux { @@ -78,8 +87,12 @@ class StreamHead( fun notify(chain: Chain, client: TopicProcessor) { val upstream = upstreams.getUpstream(chain) ?: return val head = upstream.getHead().getHead() - head.subscribe { - notify(chain, it, client) + head.subscribe { block -> + UpstreamServices.onceOk(upstream).subscribe { avail -> + if (avail) { + notify(chain, block, client) + } + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 24338354..cdf3d019 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -18,6 +18,9 @@ class ChainUpstreams ( init { head = updateHead() + observeStatus() + .distinctUntilChanged() + .subscribe { printStatus() } } internal fun updateHead(): EthereumHead { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt index 61bf1f03..ace32ec4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt @@ -42,11 +42,9 @@ class EthereumRpcHead( curr == null || curr.totalDifficulty < block.totalDifficulty } .subscribe { block -> + head.set(block) stream.onNext(block) } - - Flux.from(this.stream) - .subscribe { head.set(it) } } override fun getHead(): Mono> { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt index 76dbb015..2790930e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt @@ -32,10 +32,9 @@ class EthereumWs( return } client.onNewBlock { + head.set(it) topic.onNext(it) } - - getFlux().subscribe { head.set(it) } } fun getFlux(): Flux> { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index 6f4f04c0..92da29b8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -89,9 +89,9 @@ open class GrpcUpstream( } .subscribe { block -> log.debug("New block ${block.number} on ${chain}") + setStatus(UpstreamAvailability.OK) headBlock.set(block) streamBlocks.onNext(block) - setStatus(UpstreamAvailability.OK) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt new file mode 100644 index 00000000..a87265c4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamServices.kt @@ -0,0 +1,18 @@ +package io.emeraldpay.dshackle.upstream + +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import java.time.Duration + +class UpstreamServices { + + companion object { + fun onceOk(up: Upstream, waitTime: Duration = Duration.ofSeconds(15)): Mono { + return Flux.concat(Flux.just(up.getStatus()), up.observeStatus()) + .timeout(waitTime, Mono.just(UpstreamAvailability.UNAVAILABLE)) + .filter { it == UpstreamAvailability.OK } + .next() + .hasElement() + } + } +} \ No newline at end of file