diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 06625c45..8ce07355 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -59,10 +59,12 @@ class ChainUpstreams ( upstream.setLag(0) upstream.getHead() } else { - val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }) - newHead.start() - val lagObserver = HeadLagObserver(newHead, upstreams) - lagObserver.start() + val newHead = EthereumHeadMerge(upstreams.map { it.getHead().getFlux() }).apply { + this.start() + } + val lagObserver = HeadLagObserver(newHead, upstreams).apply { + this.start() + } this.lagObserver = lagObserver newHead } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt index 8ec95b18..3686cee4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson +import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable @@ -11,7 +12,7 @@ import java.io.Closeable import java.util.concurrent.atomic.AtomicReference class EthereumHeadMerge( - upstreams: List + fluxes: Iterable>> ): EthereumHead, Lifecycle { private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java) @@ -20,7 +21,6 @@ class EthereumHeadMerge( private var subscription: Disposable? = null init { - val fluxes = upstreams.map { it.getFlux() } flux = Flux.merge(fluxes) .distinctUntilChanged { it.hash diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 83a3a572..d57bf5e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -5,7 +5,6 @@ import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable -import java.io.Closeable open class EthereumUpstream( val chain: Chain, @@ -57,7 +56,17 @@ open class EthereumUpstream( open fun createHead(): EthereumHead { return if (ethereumWs != null) { - EthereumWsHead(ethereumWs).apply { + // load current block through RPC then listen for following blocks through WS + val ws = EthereumWsHead(ethereumWs).apply { + this.start() + } + val rpc = EthereumRpcHead(api).apply { + this.start() + } + val currentHead = rpc.getHead().doFinally { + rpc.stop() + } + EthereumHeadMerge(listOf(currentHead, ws.getFlux())).apply { this.start() } } else {