solution: subscribe to new blocks through WebSockets

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'

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<Any>): Mono<ByteArray> {
         val result: Mono<Any> = if (hardcodedMethods.contains(method)) {
-            Mono.just(method)
+            Mono.just(method).map { hardcoded(it) }
-                .map{ hardcoded(it) }
         } else if (allowedMethods.contains(method)) {
-            Mono.fromCompletionStage(
+            callUpstream(method, params)
-                        rpcClient.execute(RpcCall.create(method, Any::class.java, params))
-                )
-                .timeout(timeout)
         } else {
             Mono.error(RpcException(-32601, "Method not allowed or not found"))
         }
@@ -95,6 +96,18 @@ class EthereumUpstream(
                 }
     }
 
+    private fun callUpstream(method: String, params: List<Any>): Mono<Any> {
+        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) {

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<BlockJson<TransactionId>>()
+            .name("new-blocks")
+            .build()
+    private val head = AtomicReference<BlockJson<TransactionId>>(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<BlockJson<TransactionId>> {
+        return this.topic
+    }
+
+    fun getHead(): BlockJson<TransactionId>? {
+        return head.get()
+    }
+}

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
     }

src/main/resources/application.properties

@@ -1,2 +1,4 @@
 upstream.ethereumclassic=http://localhost:8545
-upstream.ethereum=http://localhost:8546
+upstream.ethereumclassic.ws=ws://localhost:8546
+
+upstream.ethereum=http://localhost:8645