problem: a Websocket based upstream lacks current height until it loaded through Websocket

solution: load initial block through RPC, then continue with Websocket
This commit is contained in:
Igor Artamonov
2019-08-16 21:31:06 -04:00
parent 8d83935672
commit 303d7e1164
3 changed files with 19 additions and 8 deletions

View File

@@ -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
}

View File

@@ -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<EthereumHead>
fluxes: Iterable<Publisher<BlockJson<TransactionId>>>
): 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

View File

@@ -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 {