From 974a6d7d440b1e4d25b9d771cd251d08fc689ddb Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 24 May 2021 21:08:36 -0400 Subject: [PATCH] problem: lags with head info if WS connection dropped solution: reconnect to WS after a connection error --- build.gradle | 11 +- gradle.properties | 4 +- .../kotlin/io/emeraldpay/dshackle/Global.kt | 4 + .../upstream/ethereum/EthereumWsFactory.kt | 150 +++++++++++++++--- .../upstream/ethereum/EthereumWsHead.kt | 1 + .../ethereum/EthereumWsFactorySpec.groovy | 10 +- 6 files changed, 147 insertions(+), 33 deletions(-) diff --git a/build.gradle b/build.gradle index 0c2d51ca..22d45883 100644 --- a/build.gradle +++ b/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-tcnative-boringssl-static:2.0.30.Final:linux-x86_64@jar" + implementation "io.netty:netty-handler-proxy:$nettyVersion" + 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" diff --git a/gradle.properties b/gradle.properties index da138fb5..85ed838e 100644 --- a/gradle.properties +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index eb08161b..9f21ca54 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/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()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index 50e4931c..4e5c51f7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/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 - .builder() - .name("new-blocks") - .build() + private val topic = Sinks + .many() + .unicast() + .onBackpressureBuffer() + 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() - .connectTo(uri) - .origin(origin) - basicAuth?.let { auth -> - clientBuilder.basicAuth(auth.username, auth.password) - } - val client = clientBuilder.build() - try { - client.connect() - client.onNewBlock(this::onNewBlock) - } catch (e: Exception) { - log.error("Failed to connect to websocket at $uri. Error: ${e.message}") - } + connection?.dispose() + connection = null + + val subscriptionId = AtomicReference("NOTSET") + + val objectMapper = Global.objectMapper + connection = HttpClient.create() + .doOnError( + { _, t -> + log.warn("Failed to connect to $uri. Error: ${t.message}") + // going to try to reconnect later + 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() + } + msg.subscription == null -> { + // received ID for subscription + subscriptionId.set(msg.result.asText()) + log.debug("Connected to $uri") + Mono.empty() + } + 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) { - // 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 { - return Flux.from(this.topic) - .onBackpressureLatest() + return this.topic.asFlux() + } + + override fun close() { + keepConnection = false + connection?.dispose() + connection = null } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 26ad786c..4abe9284 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/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 } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy index 088945f5..998fcae8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy +++ b/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)) } }