StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: check current height on connect

Browse Source
This commit is contained in:
Igor Artamonov
2021-09-20 22:43:00 -04:00
parent b68da0f801
commit e40c62a728
3 changed files with 48 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
View File

@@ -15,9 +15,48 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.AbstractHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.etherjar.hex.HexQuantity
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
open class DefaultEthereumHead : Head, AbstractHead() {
companion object {
private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
}
fun getLatestBlock(api: Reader<JsonRpcRequest, JsonRpcResponse>): Mono<BlockContainer> {
return api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
.subscribeOn(EthereumRpcHead.scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
.flatMap {
if (it.error != null) {
Mono.error(it.error.asException(null))
} else {
val value = it.getResultAsProcessedString()
Mono.just(HexQuantity.from(value))
}
}
.flatMap {
//fetching by Block Height here, critical to use the same upstream as in previous call,
//b/c different upstreams may have different blocks on the same height
api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
.subscribeOn(EthereumRpcHead.scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
}
.map {
BlockContainer.fromEthereumJson(it.getResult())
}
.onErrorResume { err ->
log.debug("Failed to fetch latest block: ${err.message}")
Mono.empty()
}
}
}

25
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
View File

@@ -49,30 +49,7 @@ class EthereumRpcHead(
val base = Flux.interval(interval)
.publishOn(scheduler)
.flatMap {
api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
.subscribeOn(scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
.flatMap {
if (it.error != null) {
Mono.error(it.error.asException(null))
} else {
val value = it.getResultAsProcessedString()
Mono.just(HexQuantity.from(value))
}
}
}
.flatMap {
//fetching by Block Height here, critical to use same upstream,
//different upstreams may have different blocks on the same height
api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
.subscribeOn(scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
}
.map {
BlockContainer.fromEthereumJson(it.getResult())
}
.onErrorContinue { err, _ ->
log.debug("RPC error ${err.message}")
getLatestBlock(api)
}
refreshSubscription = super.follow(base)
}

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -16,9 +16,11 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Flux
class EthereumWsHead(
private val ws: EthereumWsFactory.EthereumWs
@@ -34,7 +36,12 @@ class EthereumWsHead(
override fun start() {
this.subscription?.dispose()
this.subscription = super.follow(ws.getBlocksFlux())
val heads = Flux.merge(
// get the current block, not just wait for the next update
getLatestBlock(JsonRpcWsClient(ws)),
ws.getBlocksFlux()
)
this.subscription = super.follow(heads)
}
override fun stop() {