problem: lags with head info if WS connection dropped

solution: reconnect to WS after a connection error

build.gradle

@@ -70,8 +70,13 @@ dependencies {
     implementation "io.netty:netty-transport:$nettyVersion"
     implementation "io.netty:netty-common:$nettyVersion"
     implementation "io.netty:netty-handler:$nettyVersion"
-    implementation "io.netty:netty-tcnative:2.0.30.Final:linux-x86_64@jar"
+    implementation "io.netty:netty-handler-proxy:$nettyVersion"
-    implementation "io.netty:netty-tcnative-boringssl-static:2.0.30.Final:linux-x86_64@jar"
+    implementation "io.netty:netty-codec:$nettyVersion"
+    implementation "io.netty:netty-codec-http2:$nettyVersion"
+    implementation "io.netty:netty-codec-http:$nettyVersion"
+    implementation "io.netty:netty-buffer:$nettyVersion"
+    implementation "io.netty:netty-tcnative:2.0.39.Final:linux-x86_64@jar"
+    implementation "io.netty:netty-tcnative-boringssl-static:2.0.39.Final:linux-x86_64@jar"
 
     implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
     implementation "org.jetbrains.kotlin:kotlin-reflect"
@@ -83,7 +88,7 @@ dependencies {
     implementation "org.springframework.security:spring-security-web:$springSecurtyVersion"
     implementation "org.springframework.security:spring-security-config:$springSecurtyVersion"
     implementation "io.projectreactor:reactor-core:$reactorVersion"
-    implementation "io.projectreactor.netty:reactor-netty:1.0.6"
+    implementation "io.projectreactor.netty:reactor-netty:1.0.7"
     implementation 'io.projectreactor.addons:reactor-extra:3.4.3'
     implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.1.3'
     implementation "com.salesforce.servicelibs:reactor-grpc-stub:$reactiveGrpcVersion"

gradle.properties

@@ -5,13 +5,13 @@ protocVersion=3.9.0
 # Main Libs
 slf4jVersion=1.7.25
 jacksonVersion=2.11.0
-grpcVersion=1.25.0
+grpcVersion=1.38.0
 reactiveGrpcVersion=1.0.1
 springBootVersion=2.4.5
 springVersion=5.3.6
 springSecurtyVersion=5.4.6
 reactorVersion=3.4.5
-nettyVersion=4.1.53.Final
+nettyVersion=4.1.65.Final
 # Our Libs
 etherjarVersion=0.10.2
 # Testing

src/main/kotlin/io/emeraldpay/dshackle/Global.kt

@@ -26,6 +26,8 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspentDeserializer
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import java.text.SimpleDateFormat
 import java.util.*
+import java.util.concurrent.Executors
+import java.util.concurrent.ScheduledExecutorService
 
 class Global {
 
@@ -36,6 +38,8 @@ class Global {
         @JvmStatic
         val objectMapper: ObjectMapper = createObjectMapper()
 
+        val control: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor()
+
         private fun createObjectMapper(): ObjectMapper {
             val module = SimpleModule("EmeraldDshackle", Version(1, 0, 0, null, null, null))
             module.addSerializer(JsonRpcResponse::class.java, JsonRpcResponse.ResponseJsonSerializer())

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt

@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.data.BlockContainer
@@ -24,14 +25,23 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
-import io.infinitape.etherjar.rpc.ws.WebsocketClient
+import io.infinitape.etherjar.rpc.ws.SubscriptionJson
+import io.netty.buffer.ByteBufInputStream
+import io.netty.handler.codec.http.HttpHeaderNames
 import org.slf4j.LoggerFactory
+import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.extra.processor.TopicProcessor
+import reactor.core.publisher.Sinks
+import reactor.netty.http.client.HttpClient
+import reactor.netty.http.client.WebsocketClientSpec
 import reactor.retry.Repeat
+import java.io.InputStream
 import java.net.URI
 import java.time.Duration
+import java.util.*
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
 
 class EthereumWsFactory(
         private val uri: URI,
@@ -49,36 +59,119 @@ class EthereumWsFactory(
             private val origin: URI,
             private val upstream: EthereumUpstream,
             private val basicAuth: AuthConfig.ClientBasicAuth?
-    ) {
+    ) : AutoCloseable {
 
         companion object {
             private val log = LoggerFactory.getLogger(EthereumWs::class.java)
+
+            private const val START_REQUEST = "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}"
         }
 
-        private val topic = TopicProcessor
+        private val topic = Sinks
-                .builder<BlockContainer>()
+                .many()
-                .name("new-blocks")
+                .unicast()
-                .build()
+                .onBackpressureBuffer<BlockContainer>()
+        private var keepConnection = true
+        private var connection: Disposable? = null
 
         fun connect() {
+            if (keepConnection) {
+                connectInternal()
+            }
+        }
+
+        private fun tryReconnectLater() {
+            Global.control.schedule(
+                    { connectInternal() },
+                    Defaults.retryConnection.seconds, TimeUnit.SECONDS)
+        }
+
+        private fun connectInternal() {
             log.info("Connecting to WebSocket: $uri")
-            val clientBuilder = WebsocketClient.newBuilder()
+            connection?.dispose()
-                    .connectTo(uri)
+            connection = null
-                    .origin(origin)
+
-            basicAuth?.let { auth ->
+            val subscriptionId = AtomicReference<String>("NOTSET")
-                clientBuilder.basicAuth(auth.username, auth.password)
+
-            }
+            val objectMapper = Global.objectMapper
-            val client = clientBuilder.build()
+            connection = HttpClient.create()
-            try {
+                    .doOnError(
-                client.connect()
+                            { _, t ->
-                client.onNewBlock(this::onNewBlock)
+                                log.warn("Failed to connect to $uri. Error: ${t.message}")
-            } catch (e: Exception) {
+                                // going to try to reconnect later
-                log.error("Failed to connect to websocket at $uri. Error: ${e.message}")
+                                tryReconnectLater()
-            }
+                            },
+                            { _, _ ->
+
+                            }
+                    )
+                    .headers { headers ->
+                        headers.add(HttpHeaderNames.ORIGIN, origin)
+                        basicAuth?.let { auth ->
+                            val tmp: String = auth.username + ":" + auth.password
+                            val base64password = Base64.getEncoder().encodeToString(tmp.toByteArray())
+                            headers.add(HttpHeaderNames.AUTHORIZATION, "Basic $base64password")
+                        }
+                    }
+                    .let {
+                        if (uri.scheme == "wss") {
+                            it.secure()
+                        } else {
+                            it
+                        }
+                    }
+                    .websocket(
+                            WebsocketClientSpec.builder()
+                                    .handlePing(true)
+                                    .compress(false)
+                                    .build()
+                    )
+
+                    .uri(uri)
+                    .handle { inbound, outbound ->
+                        val consumer = inbound.aggregateFrames()
+                                .aggregateFrames(8 * 65_536)
+                                .receiveFrames()
+                                .flatMap {
+                                    val msg: SubscriptionJson = objectMapper.readerFor(SubscriptionJson::class.java)
+                                            .readValue(ByteBufInputStream(it.content()) as InputStream)
+                                    when {
+                                        msg.error != null -> {
+                                            Mono.error(IllegalStateException("Received error from WS upstream"))
+                                        }
+                                        msg.subscription == subscriptionId.get() -> {
+                                            onNewBlock(msg.blockResult)
+                                            Mono.empty<Int>()
+                                        }
+                                        msg.subscription == null -> {
+                                            // received ID for subscription
+                                            subscriptionId.set(msg.result.asText())
+                                            log.debug("Connected to $uri")
+                                            Mono.empty<Int>()
+                                        }
+                                        else -> {
+                                            Mono.error(IllegalStateException("Unknown message received: ${msg.subscription}"))
+                                        }
+                                    }
+                                }
+                                .onErrorResume { t ->
+                                    log.warn("Connection dropped to $uri. Error: ${t.message}")
+                                    // going to try to reconnect later
+                                    tryReconnectLater()
+                                    // completes current outbound flow
+                                    Mono.empty()
+                                }
+
+
+                        outbound.sendString(Mono.just(START_REQUEST).doOnError {
+                            println("!!!!!!!")
+                        })
+                                .then(consumer.then())
+                    }.subscribe()
         }
 
         fun onNewBlock(block: BlockJson<TransactionRefJson>) {
-            // WS returns incomplete blocks
+            // WS returns incomplete blocks, i.e. without some fields, so need to fetch full block data
             if (block.difficulty == null || block.transactions == null) {
                 Mono.just(block.hash)
                         .flatMap { hash ->
@@ -100,16 +193,23 @@ class EthereumWsFactory(
                         }
                         .timeout(Defaults.timeout, Mono.empty())
                         .onErrorResume { Mono.empty() }
-                        .subscribe(topic::onNext)
+                        .subscribe {
+                            topic.tryEmitNext(it)
+                        }
 
             } else {
-                topic.onNext(BlockContainer.from(block))
+                topic.tryEmitNext(BlockContainer.from(block))
             }
         }
 
         fun getFlux(): Flux<BlockContainer> {
-            return Flux.from(this.topic)
+            return this.topic.asFlux()
-                    .onBackpressureLatest()
+        }
+
+        override fun close() {
+            keepConnection = false
+            connection?.dispose()
+            connection = null
         }
     }
 

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

@@ -38,6 +38,7 @@ class EthereumWsHead(
     }
 
     override fun stop() {
+        ws.close()
         subscription?.dispose()
         subscription = null
     }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.test.TestingCommons
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import reactor.core.publisher.Flux
 import reactor.test.StepVerifier
 import spock.lang.Specification
 
@@ -50,12 +51,15 @@ class EthereumWsFactorySpec extends Specification {
         apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
 
         when:
-        ws.onNewBlock(block)
+        def act = Flux.from(ws.getFlux())
+        new Thread({
+            ws.onNewBlock(block)
+        }).run()
 
         then:
-        StepVerifier.create(ws.flux.take(1))
+        StepVerifier.create(act)
                 .expectNext(BlockContainer.from(block))
-                .expectComplete()
+                .thenCancel()
                 .verify(Duration.ofSeconds(1))
     }
 }