diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 59afd3a7..a5bd308c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index ecd46a66..999761d5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -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): 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, client: StreamSender): Boolean { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt similarity index 93% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index 9ed57672..3de76481 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -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 { + return rpcClient.execute(batch) + } + fun execute(id: Int, method: String, params: List): Mono { val result: Mono = if (hardcodedMethods.contains(method)) { Mono.just(method).map { hardcoded(it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHead.kt new file mode 100644 index 00000000..816a3c69 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHead.kt @@ -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> { +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt new file mode 100644 index 00000000..e6a83ebc --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt @@ -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>(null) + private val stream: TopicProcessor> = 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> { + val current = head.get() + if (current != null) { + return Mono.just(current) + } + return Mono.from(stream) + } + + override fun getFlux(): Flux> { + return Flux.from(stream) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt similarity index 92% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt index 0c836d94..714c2541 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt @@ -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>() .name("new-blocks") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt new file mode 100644 index 00000000..0cb71a68 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt @@ -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>(null) + private val stream: Flux> = ws.getFlux() + + override fun getHead(): Mono> { + val current = head.get() + if (current != null) { + return Mono.just(current) + } + return Mono.from(stream) + } + + override fun getFlux(): Flux> { + return Flux.from(stream) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt new file mode 100644 index 00000000..a03574eb --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -0,0 +1,11 @@ +package io.emeraldpay.dshackle.upstream + +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +interface Head { + + fun getHead(): Mono + + fun getFlux(): Flux +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt new file mode 100644 index 00000000..82116595 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -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 { + + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index f8165f45..ec9a5761 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -17,34 +17,39 @@ class Upstreams( ) { private var seq = 0 - private val chainMapping = HashMap>() + private val chainMapping = HashMap() @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] } } \ No newline at end of file