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

solution: update current head through RPC as well

Browse Source
This commit is contained in:
Igor Artamonov
2019-06-10 18:52:32 -04:00
parent 59ab8b9010
commit e06ce56742
10 changed files with 171 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -30,7 +30,7 @@ class NativeCall(
if (chain == Chain.UNSPECIFIED) {
throw Exception("Invalid chain id: ${request.chain.number}")
}
val upstream = upstreams.ethereumUpstream(chain) ?: throw Exception("Chain ${chain.id} is unavailable")
val upstream = upstreams.ethereumUpstream(chain)?.api ?: throw Exception("Chain ${chain.id} is unavailable")
request.itemsList.toFlux()
.map {
val method = it.target

11
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -27,7 +27,7 @@ class StreamHead(
@PostConstruct
fun init() {
listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.MORDEN).forEach { chain ->
if (upstreams.ethereumUpstream(chain)?.ws != null) {
if (upstreams.ethereumUpstream(chain)?.head != null) {
clients[chain] = ConcurrentLinkedQueue()
subscribe(chain)
}
@@ -35,7 +35,7 @@ class StreamHead(
}
private fun subscribe(chain: Chain) {
upstreams.ethereumUpstream(chain)!!.ws!!.getFlux()
upstreams.ethereumUpstream(chain)!!.head.getFlux()
.doOnComplete {
log.info("Closing streams for ${chain.chainCode}")
clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client ->
@@ -68,9 +68,10 @@ class StreamHead(
fun process(chain: Chain, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
val upstream = upstreams.ethereumUpstream(chain) ?: return false
val ws = upstream.ws ?: return false
val head = ws.getHead() ?: return false
return notify(chain, head, client)
val head = upstream.head.getHead()
return head.map {
notify(chain, it, client)
}.defaultIfEmpty(false).block()!!
}
fun notify(chain: Chain, block: BlockJson<TransactionId>, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt
View File

@@ -3,24 +3,27 @@ package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.RpcCall
import io.infinitape.etherjar.rpc.RpcClient
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.json.ResponseJson
import io.infinitape.etherjar.rpc.transport.BatchStatus
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.time.Duration
import java.util.*
import java.util.concurrent.CompletableFuture
class EthereumUpstream(
class EthereumApi(
private val rpcClient: RpcClient,
private val objectMapper: ObjectMapper,
private val chain: Chain
) {
private val timeout = Duration.ofSeconds(5)
private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
var ws: EthereumWsUpstream? = null
private val log = LoggerFactory.getLogger(EthereumApi::class.java)
var ws: EthereumWs? = null
set(value) {
field = value
}
@@ -62,6 +65,10 @@ class EthereumUpstream(
"eth_accounts"
)
fun execute(batch: Batch): CompletableFuture<BatchStatus> {
return rpcClient.execute(batch)
}
fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
val result: Mono<Any> = if (hardcodedMethods.contains(method)) {
Mono.just(method).map { hardcoded(it) }

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHead.kt Normal file
View File

@@ -0,0 +1,7 @@
package io.emeraldpay.dshackle.upstream
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
interface EthereumHead: Head<BlockJson<TransactionId>> {
}

64
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt Normal file
View File

@@ -0,0 +1,64 @@
package io.emeraldpay.dshackle.upstream
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.TopicProcessor
import java.time.Duration
import java.util.concurrent.atomic.AtomicReference
class EthereumRpcHead(
private val api: EthereumApi
): EthereumHead {
private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java)
private val head = AtomicReference<BlockJson<TransactionId>>(null)
private val stream: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
fun start() {
Flux.interval(Duration.ofSeconds(7))
.flatMap {
val batch = Batch()
val f = batch.add(Commands.eth().blockNumber)
api.execute(batch)
Mono.fromCompletionStage(f)
}
.flatMap {
val batch = Batch()
val f = batch.add(Commands.eth().getBlock(it))
api.execute(batch)
Mono.fromCompletionStage(f)
}
.onErrorContinue { err, _ ->
log.warn("RPC error ${err.message}")
}
.filter { block ->
val curr = head.get()
curr == null || curr.difficulty < block.difficulty
}
.subscribe { block ->
stream.onNext(block)
}
Flux.from(this.stream)
.subscribe { head.set(it) }
}
override fun getHead(): Mono<BlockJson<TransactionId>> {
val current = head.get()
if (current != null) {
return Mono.just(current)
}
return Mono.from(stream)
}
override fun getFlux(): Flux<BlockJson<TransactionId>> {
return Flux.from(stream)
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt
View File

@@ -10,12 +10,12 @@ import java.net.URI
import java.time.Duration
import java.util.concurrent.atomic.AtomicReference
class EthereumWsUpstream(
class EthereumWs(
private val uri: URI,
private val origin: URI
) {
private val log = LoggerFactory.getLogger(EthereumWsUpstream::class.java)
private val log = LoggerFactory.getLogger(EthereumWs::class.java)
private val topic = TopicProcessor
.builder<BlockJson<TransactionId>>()
.name("new-blocks")

28
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt Normal file
View File

@@ -0,0 +1,28 @@
package io.emeraldpay.dshackle.upstream
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.concurrent.atomic.AtomicReference
class EthereumWsHead(
private val ws: EthereumWs
): EthereumHead {
private val head = AtomicReference<BlockJson<TransactionId>>(null)
private val stream: Flux<BlockJson<TransactionId>> = ws.getFlux()
override fun getHead(): Mono<BlockJson<TransactionId>> {
val current = head.get()
if (current != null) {
return Mono.just(current)
}
return Mono.from(stream)
}
override fun getFlux(): Flux<BlockJson<TransactionId>> {
return Flux.from(stream)
}
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt Normal file
View File

@@ -0,0 +1,11 @@
package io.emeraldpay.dshackle.upstream
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
interface Head<T> {
fun getHead(): Mono<T>
fun getFlux(): Flux<T>
}

24
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt Normal file
View File

@@ -0,0 +1,24 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
class Upstream(
val chain: Chain,
val api: EthereumApi,
private val ethereumWs: EthereumWs? = null
) {
val head: EthereumHead = if (ethereumWs != null) {
EthereumWsHead(ethereumWs)
} else {
EthereumRpcHead(api).apply {
this.start()
}
}
init {
}
}

36
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt
View File

@@ -17,34 +17,39 @@ class Upstreams(
) {
private var seq = 0
private val chainMapping = HashMap<Chain, List<EthereumUpstream>>()
private val chainMapping = HashMap<Chain, Upstream>()
@PostConstruct
fun start() {
env.getProperty("upstream.ethereum")?.let {
chainMapping[Chain.ETHEREUM] = listOf(buildClient(it, Chain.ETHEREUM))
val api = buildClient(it, Chain.ETHEREUM)
chainMapping[Chain.ETHEREUM] = Upstream(Chain.ETHEREUM, api)
}
env.getProperty("upstream.ethereumclassic")?.let {
chainMapping[Chain.ETHEREUM_CLASSIC] = listOf(buildClient(it, Chain.ETHEREUM_CLASSIC))
}
env.getProperty("upstream.ethereumclassic.ws")?.let {
chainMapping[Chain.ETHEREUM_CLASSIC]!![0].ws = buildWs(it, Chain.ETHEREUM_CLASSIC)
val api = buildClient(it, Chain.ETHEREUM_CLASSIC)
val ws = if (env.containsProperty("upstream.ethereumclassic.ws")) {
buildWs(env.getProperty("upstream.ethereumclassic.ws")!!, Chain.ETHEREUM_CLASSIC)
} else {
null
}
chainMapping[Chain.ETHEREUM_CLASSIC] = Upstream(Chain.ETHEREUM_CLASSIC, api, ws)
}
env.getProperty("upstream.morden")?.let {
chainMapping[Chain.MORDEN] = listOf(buildClient(it, Chain.MORDEN))
val api = buildClient(it, Chain.MORDEN)
chainMapping[Chain.MORDEN] = Upstream(Chain.MORDEN, api)
}
}
private fun buildClient(url: String, chain: Chain): EthereumUpstream {
return EthereumUpstream(
private fun buildClient(url: String, chain: Chain): EthereumApi {
return EthereumApi(
DefaultRpcClient(DefaultRpcTransport(URI(url))),
objectMapper,
chain
)
}
private fun buildWs(url: String, chain: Chain): EthereumWsUpstream {
val ws = EthereumWsUpstream(
private fun buildWs(url: String, chain: Chain): EthereumWs {
val ws = EthereumWs(
URI(url),
URI("http://localhost")
)
@@ -52,12 +57,7 @@ class Upstreams(
return ws
}
fun validateUpstream(upstream: EthereumUpstream): Boolean {
return true
}
fun ethereumUpstream(chain: Chain): EthereumUpstream? {
val all = chainMapping[chain] ?: return null
return all[seq++ % all.size]
fun ethereumUpstream(chain: Chain): Upstream? {
return chainMapping[chain]
}
}