From 4a87047b7ad359994ce5db25a23f0ef45933ef79 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 9 Jun 2019 19:59:23 -0400 Subject: [PATCH] solution: subscribe to new blocks through WebSockets --- build.gradle | 2 + .../dshackle/upstream/EthereumUpstream.kt | 25 +++++++--- .../dshackle/upstream/EthereumWsUpstream.kt | 46 +++++++++++++++++++ .../emeraldpay/dshackle/upstream/Upstreams.kt | 12 +++++ src/main/resources/application.properties | 4 +- 5 files changed, 82 insertions(+), 7 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt diff --git a/build.gradle b/build.gradle index 7cc53610..4e9dfda0 100644 --- a/build.gradle +++ b/build.gradle @@ -48,6 +48,7 @@ dependencies { compile "io.grpc:grpc-stub:${grpcVersion}" compile "io.grpc:grpc-netty:${grpcVersion}" compile "io.netty:netty-tcnative-boringssl-static:2.0.22.Final" + compile "io.netty:netty-all:4.1.36.Final" compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8" compile "org.jetbrains.kotlin:kotlin-reflect" @@ -65,6 +66,7 @@ dependencies { compile "io.infinitape:etherjar-domain:$etherjarVersion" compile "io.infinitape:etherjar-hex:$etherjarVersion" compile "io.infinitape:etherjar-rpc-http:$etherjarVersion" + compile "io.infinitape:etherjar-rpc-ws:$etherjarVersion" compile "io.infinitape:etherjar-tx:$etherjarVersion" compile 'org.apache.httpcomponents:httpmime:4.5.8' diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 9a19ac6b..9ed57672 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -2,6 +2,7 @@ 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.RpcCall import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.RpcException @@ -19,6 +20,10 @@ class EthereumUpstream( private val timeout = Duration.ofSeconds(5) private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) + var ws: EthereumWsUpstream? = null + set(value) { + field = value + } private val allowedMethods = listOf( "eth_gasPrice", @@ -59,13 +64,9 @@ class EthereumUpstream( fun execute(id: Int, method: String, params: List): Mono { val result: Mono = if (hardcodedMethods.contains(method)) { - Mono.just(method) - .map{ hardcoded(it) } + Mono.just(method).map { hardcoded(it) } } else if (allowedMethods.contains(method)) { - Mono.fromCompletionStage( - rpcClient.execute(RpcCall.create(method, Any::class.java, params)) - ) - .timeout(timeout) + callUpstream(method, params) } else { Mono.error(RpcException(-32601, "Method not allowed or not found")) } @@ -95,6 +96,18 @@ class EthereumUpstream( } } + private fun callUpstream(method: String, params: List): Mono { + if (ws != null && method == "eth_blockNumber") { + val head = ws!!.getHead() + if (head != null) { + return Mono.just(HexQuantity.from(head.number).toHex()) + } + } + return Mono.fromCompletionStage( + rpcClient.execute(RpcCall.create(method, Any::class.java, params)) + ).timeout(timeout) + } + fun hardcoded(method: String): Any { if ("net_version" == method) { if (Chain.ETHEREUM == chain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt new file mode 100644 index 00000000..0f60a080 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt @@ -0,0 +1,46 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.ws.WebsocketClient +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.TopicProcessor +import java.net.URI +import java.time.Duration +import java.util.concurrent.atomic.AtomicReference + +class EthereumWsUpstream( + private val uri: URI, + private val origin: URI +) { + + private val log = LoggerFactory.getLogger(EthereumWsUpstream::class.java) + private val topic = TopicProcessor + .builder>() + .name("new-blocks") + .build() + private val head = AtomicReference>(null); + + fun connect() { + log.info("Connecting to WebSocket: $uri") + val client = WebsocketClient() + client.connect(uri, origin) + client.onNewBlock { + topic.onNext(it) + } + + topic + .onBackpressureLatest() + .sample(Duration.ofMillis(100)) + .subscribe { head.set(it) } + } + + fun getFlux(): Flux> { + return this.topic + } + + fun getHead(): BlockJson? { + return head.get() + } +} \ 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 592a3535..f8165f45 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -27,6 +27,9 @@ class Upstreams( 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) + } env.getProperty("upstream.morden")?.let { chainMapping[Chain.MORDEN] = listOf(buildClient(it, Chain.MORDEN)) } @@ -40,6 +43,15 @@ class Upstreams( ) } + private fun buildWs(url: String, chain: Chain): EthereumWsUpstream { + val ws = EthereumWsUpstream( + URI(url), + URI("http://localhost") + ) + ws.connect() + return ws + } + fun validateUpstream(upstream: EthereumUpstream): Boolean { return true } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 5607da14..322059df 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -1,2 +1,4 @@ upstream.ethereumclassic=http://localhost:8545 -upstream.ethereum=http://localhost:8546 \ No newline at end of file +upstream.ethereumclassic.ws=ws://localhost:8546 + +upstream.ethereum=http://localhost:8645 \ No newline at end of file