diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 0fb7c3a6..addd605d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -76,6 +76,7 @@ class UpstreamsConfig { class WsEndpoint(val url: URI) { var origin: URI? = null + var basicAuth: BasicAuth? = null } open class Auth { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index a4041596..5eb58f39 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -60,7 +60,7 @@ class UpstreamsConfigReader { getValueAsString(node, "origin")?.let { origin -> ws.origin = URI(origin) } -// ws.auth = readAuth(getMapping(node, "auth")) + ws.basicAuth = readBasicAuth(node) } } } else if (hasAny(connNode, "grpc")) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index f7295c94..adf6007d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -105,7 +105,6 @@ open class ConfiguredUpstreams( options: UpstreamsConfig.Options, labels: UpstreamsConfig.Labels) { var rpcApi: EthereumApi? = null - var wsApi: EthereumWs? = null val urls = ArrayList() up.rpc?.let { endpoint -> val rpcTransport = DefaultRpcTransport(endpoint.url) @@ -126,15 +125,21 @@ open class ConfiguredUpstreams( ) urls.add(endpoint.url) } - up.ws?.let { endpoint -> - wsApi = EthereumWs( - endpoint.url, - endpoint.origin ?: URI("http://localhost") - ) - wsApi!!.connect() - urls.add(endpoint.url) - } if (rpcApi != null) { + val wsApi: EthereumWs? = up.ws?.let { endpoint -> + val wsApi = EthereumWs( + endpoint.url, + endpoint.origin ?: URI("http://localhost"), + rpcApi!! + ) + endpoint.basicAuth?.let { auth -> + wsApi.basicAuth = auth + } + wsApi.connect() + urls.add(endpoint.url) + wsApi + } + log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain)) ethereumUpstream.start() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index 043e9348..6b8717e3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -38,13 +38,10 @@ open class EthereumApi( } open fun execute(id: Int, method: String, params: List): Mono { - val result: Mono = if (targets.isHardcoded(method)) { - Mono.just(method) - .map { targets.hardcoded(it) } - } else if (targets.isAllowed(method)) { - callUpstream(method, params) - } else { - Mono.error(RpcException(-32601, "Method not allowed or not found")) + val result: Mono = when { + targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) } + targets.isAllowed(method) -> callUpstream(method, params) + else -> Mono.error(RpcException(-32601, "Method not allowed or not found")) } return result .doOnError { t -> @@ -72,11 +69,13 @@ open class EthereumApi( } } - 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()) + private fun callUpstream(method: String, params: List): Mono { + if (method == "eth_blockNumber") { + val current = upstream?.getHead()?.getHead()?.let { head -> + head.map { HexQuantity.from(it.number).toHex() } + } + if (current != null) { + return current } } return Mono.fromCompletionStage( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 38edf221..83a3a572 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -57,7 +57,9 @@ open class EthereumUpstream( open fun createHead(): EthereumHead { return if (ethereumWs != null) { - EthereumWsHead(ethereumWs) + EthereumWsHead(ethereumWs).apply { + this.start() + } } else { EthereumRpcHead(api).apply { this.start() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt index 2790930e..0ebc7471 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt @@ -1,18 +1,22 @@ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.Commands 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.Mono import reactor.core.publisher.TopicProcessor +import reactor.retry.Repeat import java.net.URI import java.time.Duration -import java.util.concurrent.atomic.AtomicReference class EthereumWs( private val uri: URI, - private val origin: URI + private val origin: URI, + private val api: EthereumApi ) { private val log = LoggerFactory.getLogger(EthereumWs::class.java) @@ -20,20 +24,34 @@ class EthereumWs( .builder>() .name("new-blocks") .build() - private val head = AtomicReference>(null) + var basicAuth: UpstreamsConfig.BasicAuth? = null fun connect() { log.info("Connecting to WebSocket: $uri") - val client = WebsocketClient() + val client = WebsocketClient(uri, origin) + basicAuth?.let { auth -> + client.setBasicAuth(auth.username, auth.password) + } try { - client.connect(uri, origin) + client.connect() } catch (e: Exception) { log.error("Failed to connect to websocket at $uri. Error: ${e.message}") return } client.onNewBlock { - head.set(it) - topic.onNext(it) + if (it.totalDifficulty == null || it.transactions == null) { + Mono.just(it.hash).flatMap { hash -> + api.executeAndConvert(Commands.eth().getBlock(hash)) + }.repeatWhenEmpty { n -> + Repeat.times(10) + .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250)) + .apply(n) + } + .timeout(Duration.ofSeconds(5), Mono.empty()) + .subscribe(topic::onNext) + } else { + topic.onNext(it) + } } } @@ -42,8 +60,4 @@ class EthereumWs( .onBackpressureLatest() .sample(Duration.ofMillis(100)) } - - fun getHead(): BlockJson? { - return head.get() - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt index 522b5523..f557fc79 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt @@ -1,28 +1,56 @@ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.util.concurrent.atomic.AtomicReference class EthereumWsHead( private val ws: EthereumWs -): EthereumHead { +): EthereumHead, Lifecycle { + private val log = LoggerFactory.getLogger(EthereumWsHead::class.java) + + private var subscription: Disposable? = null private val head = AtomicReference>(null) - private val stream: Flux> = ws.getFlux() + private var stream: Flux>? = null override fun getHead(): Mono> { val current = head.get() if (current != null) { return Mono.just(current) } - return Mono.from(stream) + return Mono.from(getFlux()) } override fun getFlux(): Flux> { - return ws.getFlux() + return stream?.let { Flux.from(it) } ?: Flux.error(Exception("Not started")) + } + + override fun isRunning(): Boolean { + return subscription != null + } + + override fun start() { + val flux = ws.getFlux() + .distinctUntilChanged { it.hash } + .filter { block -> + val curr = head.get() + curr == null || curr.totalDifficulty < block.totalDifficulty + }.share() + + this.subscription = flux.subscribe(head::set) + this.stream = flux + } + + override fun stop() { + subscription?.dispose() + subscription = null } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index d9fad900..0dbef1ff 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -31,6 +31,11 @@ class UpstreamsConfigReaderSpec extends Specification { rpc.url == new URI("http://localhost:8545") ws != null ws.url == new URI("ws://localhost:8546") + ws.basicAuth != null + with(ws.basicAuth) { + username == "9c199ad8f281f20154fc258fe41a6814" + password == "258fe4149c199ad8f2811a68f20154fc" + } } } with(act.upstreams.get(1)) { diff --git a/src/test/resources/upstreams-basic.yaml b/src/test/resources/upstreams-basic.yaml index ce7a478c..0198d513 100644 --- a/src/test/resources/upstreams-basic.yaml +++ b/src/test/resources/upstreams-basic.yaml @@ -16,6 +16,9 @@ upstreams: ws: url: "ws://localhost:8546" origin: "http://localhost" + basic-auth: + username: 9c199ad8f281f20154fc258fe41a6814 + password: 258fe4149c199ad8f2811a68f20154fc - id: infura chain: ethereum connection: