diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 83af5646..b13d9a02 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -106,7 +106,7 @@ class UpstreamsConfigReader( config.upstreams.add(upstream) val connection = UpstreamsConfig.GrpcConnection() upstream.connection = connection - getValueAsInt(connConfigNode, "node-rating")?.let { + getValueAsInt(connConfigNode, "upstream-rating")?.let { connection.upstreamRating = it } getValueAsString(connConfigNode, "host")?.let { @@ -170,7 +170,7 @@ class UpstreamsConfigReader( getMapping(connConfigNode, "execution")?.let { connection.execution = readEthereumConnection(it) } - getValueAsInt(connConfigNode, "node-rating")?.let { + getValueAsInt(connConfigNode, "upstream-rating")?.let { connection.upstreamRating = it } return connection diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 5097deee..819527ff 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -159,7 +159,7 @@ open class ConfiguredUpstreams( return null } val urls = ArrayList() - val connectorFactory = buildEthereumConnectorFactory(execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating)) + val connectorFactory = buildEthereumConnectorFactory(config.id!!, execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating)) val methods = buildMethods(config, chain) if (connectorFactory == null) { return null @@ -228,7 +228,7 @@ open class ConfiguredUpstreams( val urls = ArrayList() val methods = buildMethods(config, chain) - val connectorFactory = buildEthereumConnectorFactory(conn, chain, urls, MostWorkForkChoice()) + val connectorFactory = buildEthereumConnectorFactory(config.id!!, conn, chain, urls, MostWorkForkChoice()) if (connectorFactory == null) { return null } @@ -280,9 +280,10 @@ open class ConfiguredUpstreams( } } - private fun buildWsFactory(conn: UpstreamsConfig.EthereumConnection, urls: ArrayList? = null): EthereumWsFactory? { + private fun buildWsFactory(id: String, chain: Chain, conn: UpstreamsConfig.EthereumConnection, urls: ArrayList? = null): EthereumWsFactory? { return conn.ws?.let { endpoint -> val wsApi = EthereumWsFactory( + id, chain, endpoint.url, endpoint.origin ?: URI("http://localhost"), ) @@ -295,8 +296,8 @@ open class ConfiguredUpstreams( } } - private fun buildEthereumConnectorFactory(conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList, forkChoice: ForkChoice): EthereumConnectorFactory? { - val wsFactoryApi = buildWsFactory(conn, urls) + private fun buildEthereumConnectorFactory(id: String, conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList, forkChoice: ForkChoice): EthereumConnectorFactory? { + val wsFactoryApi = buildWsFactory(id, chain, conn, urls) val httpFactory = buildHttpFactory(conn, urls) log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt index aa2b623d..996d9a31 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent import io.netty.handler.codec.http.HttpHeaderNames import io.netty.handler.codec.http.HttpHeaders import io.netty.handler.ssl.SslContextBuilder +import io.netty.resolver.DefaultAddressResolverGroup import org.bitcoinj.core.Address import org.slf4j.LoggerFactory import reactor.core.publisher.Mono @@ -47,6 +48,7 @@ class EsploraClient( init { var build = HttpClient.create() + .resolver(DefaultAddressResolverGroup.INSTANCE) build = build.headers { h -> h.add(HttpHeaderNames.CONTENT_TYPE, "application/json") 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 da7aa412..00def0bf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -20,18 +20,49 @@ import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag +import io.micrometer.core.instrument.Timer import java.net.URI class EthereumWsFactory( + private val id: String, + private val chain: Chain, private val uri: URI, - private val origin: URI + private val origin: URI, ) { var basicAuth: AuthConfig.ClientBasicAuth? = null var config: UpstreamsConfig.WsEndpoint? = null - fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): WsConnection { - return WsConnection(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { ws -> + // metrics are shared between all connections to the same WS + private val metrics: RpcMetrics = run { + val metricsTags = listOf( + Tag.of("upstream", id), + // UNSPECIFIED shouldn't happen too + Tag.of("chain", chain.chainCode) + ) + + RpcMetrics( + Timer.builder("upstream.ws.conn") + .description("Request time through a WebSocket JSON RPC connection") + .tags(metricsTags) + .publishPercentileHistogram() + .register(Metrics.globalRegistry), + Counter.builder("upstream.ws.fail") + .description("Number of failures of WebSocket JSON RPC requests") + .tags(metricsTags) + .register(Metrics.globalRegistry) + ) + } + + fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?): WsConnection { + require(upstream == null || upstream.getId() == id) { + "Creating instance for different upstream. ${upstream?.getId()} != id" + } + return WsConnection(uri, origin, basicAuth, metrics, upstream, validator).also { ws -> config?.frameSize?.let { ws.frameSize = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index f5fa0159..d09cda9e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser @@ -34,6 +35,7 @@ import io.netty.buffer.ByteBuf import io.netty.buffer.ByteBufInputStream import io.netty.buffer.Unpooled import io.netty.handler.codec.http.HttpHeaderNames +import io.netty.resolver.DefaultAddressResolverGroup import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.util.backoff.BackOff @@ -53,6 +55,7 @@ import reactor.retry.Repeat import reactor.util.function.Tuples import java.net.URI import java.time.Duration +import java.time.Instant import java.util.Base64 import java.util.concurrent.Executors import java.util.concurrent.TimeUnit @@ -60,7 +63,7 @@ import java.util.concurrent.TimeoutException import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger -class WsConnection( +open class WsConnection( private val uri: URI, private val origin: URI, private val basicAuth: AuthConfig.ClientBasicAuth?, @@ -111,12 +114,19 @@ class WsConnection( .many() .multicast() .directBestEffort() + private val disconnects = Sinks + .many() + .multicast() + .directBestEffort() private val sendIdSeq = AtomicInteger(IDS_START) private val sendExecutor = Executors.newSingleThreadExecutor() private var keepConnection = true private var connection: Disposable? = null private val reconnecting = AtomicBoolean(false) + open val isConnected: Boolean + get() = connection != null && !reconnecting.get() + fun setReconnectIntervalSeconds(value: Long) { reconnectBackoff = FixedBackOff(value * 1000, FixedBackOff.UNLIMITED_ATTEMPTS) currentBackOff = reconnectBackoff.start() @@ -161,7 +171,9 @@ class WsConnection( log.info("Connecting to WebSocket: $uri") connection?.dispose() connection = HttpClient.create() + .resolver(DefaultAddressResolverGroup.INSTANCE) .doOnDisconnected { + disconnects.tryEmitNext(Instant.now()) log.info("Disconnected from $uri") // mark upstream as UNAVAIL upstream?.setStatus(UpstreamAvailability.UNAVAILABLE) @@ -371,25 +383,39 @@ class WsConnection( fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono { val expectedId = request.id.toLong() - val failResponse = JsonRpcResponse( - null, + val noResponse = JsonRpcException( + JsonRpcResponse.Id.from(originalId), JsonRpcError( RpcResponseError.CODE_INTERNAL_ERROR, "Response not received from WebSocket" - ), - JsonRpcResponse.Id.from(originalId), null + ) ) - return Flux.from(rpcReceive.asFlux()) + val response = Flux.from(rpcReceive.asFlux()) .doOnSubscribe { sendRpc(request) } .filter { resp -> resp.id.asNumber() == expectedId } .take(Defaults.timeout) .take(1) .singleOrEmpty() + + val failOnDisconnect = Mono.from(disconnects.asFlux()) + .flatMap { + Mono.error( + JsonRpcException( + JsonRpcResponse.Id.from(originalId), + JsonRpcError( + RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, + "Disconnected from WebSocket" + ) + ) + ) + } + + return response.or(failOnDisconnect) .doOnNext { rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) } .doOnError { rpcMetrics?.fails?.increment() } .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) } - .defaultIfEmpty(failResponse) + .switchIfEmpty(Mono.error(noResponse)) } fun getBlocksFlux(): Flux { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index 03c541b1..54816a74 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -32,7 +32,7 @@ class EthereumRpcConnector( init { if (wsFactory != null) { // do not set upstream to the WS, since it doesn't control the RPC upstream - conn = wsFactory.create(null, null, null) + conn = wsFactory.create(null, null) val wsHead = EthereumWsHead(conn, forkChoice) // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed val rpcHead = EthereumRpcHead(directReader, forkChoice, Duration.ofSeconds(60)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index c4844dda..369bb564 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -30,24 +30,7 @@ class EthereumWsConnector( private val head: EthereumWsHead init { - val metricsTags = listOf( - Tag.of("upstream", upstream.getId()), - // UNSPECIFIED shouldn't happen too - Tag.of("chain", chain.chainCode) - ) - val metrics = RpcMetrics( - Timer.builder("upstream.ws.conn") - .description("Request time through a WebSocket JSON RPC connection") - .tags(metricsTags) - .publishPercentileHistogram() - .register(Metrics.globalRegistry), - Counter.builder("upstream.ws.fail") - .description("Number of failures of WebSocket JSON RPC requests") - .tags(metricsTags) - .register(Metrics.globalRegistry) - ) - - conn = wsFactory.create(upstream, validator, metrics) + conn = wsFactory.create(upstream, validator) head = EthereumWsHead(conn, forkChoice) api = JsonRpcWsClient(conn) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index 8d3e7b76..fb425ebf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt @@ -23,6 +23,7 @@ import io.netty.buffer.Unpooled import io.netty.handler.codec.http.HttpHeaderNames import io.netty.handler.codec.http.HttpHeaders import io.netty.handler.ssl.SslContextBuilder +import io.netty.resolver.DefaultAddressResolverGroup import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import reactor.netty.http.client.HttpClient @@ -53,6 +54,7 @@ class JsonRpcHttpClient( init { var build = HttpClient.create() + .resolver(DefaultAddressResolverGroup.INSTANCE) build = build.headers { h -> h.add(HttpHeaderNames.CONTENT_TYPE, "application/json") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt new file mode 100644 index 00000000..82ab68cd --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt @@ -0,0 +1,27 @@ +package io.emeraldpay.dshackle.upstream.rpcclient + +import io.emeraldpay.dshackle.reader.Reader +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +/** + * An aggregating JSON RPC Client that wraps two actual readers, a Primary and a Secondary. + * It always calls the Primary reader, and if it fails or produces an empty result, then it calls the Secondary reader. + */ +class JsonRpcSwitchClient( + private val primary: Reader, + private val secondary: Reader, +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(JsonRpcSwitchClient::class.java) + } + + override fun read(key: JsonRpcRequest): Mono { + return primary.read(key) + .switchIfEmpty(Mono.error(IllegalStateException("No response from Primary Connection"))) + .onErrorResume { + secondary.read(key) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt index 0a10a868..6f65d0a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.rpcclient import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.ethereum.WsConnection +import io.emeraldpay.etherjar.rpc.RpcResponseError import reactor.core.publisher.Mono class JsonRpcWsClient( @@ -24,6 +25,17 @@ class JsonRpcWsClient( ) : Reader { override fun read(key: JsonRpcRequest): Mono { + if (!ws.isConnected) { + return Mono.error( + JsonRpcException( + JsonRpcResponse.NumberId(key.id), + JsonRpcError( + RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, + "WebSocket is not connected" + ) + ) + ) + } return ws.call(key) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 2244dddc..99b81f93 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -166,7 +166,7 @@ class UpstreamsConfigReaderSpec extends Specification { with((UpstreamsConfig.EthereumPosConnection) connection) { execution.rpc != null execution.rpc.url == new URI("http://34.106.60.110:8545") - blockPriority == 100 + upstreamRating == 100 } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy index a84e6e00..eb3f22c0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy @@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.test.MockWSServer import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.grpc.Chain import reactor.test.StepVerifier import spock.lang.Shared import spock.lang.Specification @@ -30,7 +31,7 @@ class WsConnectionRealSpec extends Specification { server = new MockWSServer(port) server.start() Thread.sleep(SLEEP) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null) + conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null) } def cleanup() { @@ -93,10 +94,26 @@ class WsConnectionRealSpec extends Specification { act[0].value.contains("\"params\":[\"newHeads\"]") } + def "Error on request when server disconnects"() { + when: + conn.connect() + conn.reconnectIntervalSeconds = 2 + + def resp = conn.call(new JsonRpcRequest("foo_bar", [])) + + then: + StepVerifier.create(resp) + .then { server.stop() } + .expectError() + .verify(Duration.ofSeconds(1)) + } + def "Gets UNAVAIL status right after disconnect"() { setup: - def up = Mock(DefaultUpstream) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null) + def up = Mock(DefaultUpstream) { + _ * getId() >> "test" + } + conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null) when: conn.connect() conn.reconnectIntervalSeconds = 10 @@ -111,7 +128,7 @@ class WsConnectionRealSpec extends Specification { def "Validates after connect"() { setup: def validator = Mock(EthereumUpstreamValidator) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null) + conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator) when: conn.connect() Thread.sleep(100) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy index 3b4b5edb..b8996287 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy @@ -25,6 +25,7 @@ import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson +import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux import reactor.test.StepVerifier import spock.lang.Specification @@ -37,7 +38,7 @@ class WsConnectionSpec extends Specification { def "Fetch block"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def block = new BlockJson() block.number = 100 @@ -53,7 +54,7 @@ class WsConnectionSpec extends Specification { def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block) @@ -71,10 +72,10 @@ class WsConnectionSpec extends Specification { def "Makes a RPC call"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) def tx = new TransactionJson().tap { hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") @@ -96,10 +97,10 @@ class WsConnectionSpec extends Specification { def "Makes a RPC call - return null"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null) @@ -119,10 +120,10 @@ class WsConnectionSpec extends Specification { def "Makes a RPC call - return error"() { setup: - def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null, null, null) + def ws = wsf.create(null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test")) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy new file mode 100644 index 00000000..883e400b --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy @@ -0,0 +1,70 @@ +package io.emeraldpay.dshackle.upstream.rpcclient + +import io.emeraldpay.dshackle.reader.Reader +import reactor.core.publisher.Mono +import spock.lang.Specification + +import java.time.Duration + +class JsonRpcSwitchClientSpec extends Specification { + + def "Uses primary response if it works"() { + setup: + def primaryCalled = false + def secondaryCalled = false + def request = new JsonRpcRequest("eth_test", []) + def response = JsonRpcResponse.ok("test".bytes, new JsonRpcResponse.NumberId(100)) + def primary = Mock(Reader) { + 1 * read(request) >> Mono.fromCallable { + primaryCalled = true + response + } + } + def secondary = Mock(Reader) { + _ * read(request) >> Mono.fromCallable { + secondaryCalled = true + response + } + } + + def client = new JsonRpcSwitchClient(primary, secondary) + + when: + def act = client.read(request).block(Duration.ofSeconds(1)) + + then: + act == response + primaryCalled + !secondaryCalled + } + + def "Uses secondary response if primary fails"() { + setup: + def primaryCalled = false + def secondaryCalled = false + def request = new JsonRpcRequest("eth_test", []) + def response = JsonRpcResponse.ok("test".bytes, new JsonRpcResponse.NumberId(100)) + def primary = Mock(Reader) { + 1 * read(request) >> Mono.fromCallable { + primaryCalled = true + throw new IllegalStateException("Primary Fail") + } + } + def secondary = Mock(Reader) { + 1 * read(request) >> Mono.fromCallable { + secondaryCalled = true + response + } + } + + def client = new JsonRpcSwitchClient(primary, secondary) + + when: + def act = client.read(request).block(Duration.ofSeconds(1)) + + then: + act == response + primaryCalled + secondaryCalled + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy new file mode 100644 index 00000000..e0ac8a1b --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy @@ -0,0 +1,23 @@ +package io.emeraldpay.dshackle.upstream.rpcclient + +import io.emeraldpay.dshackle.upstream.ethereum.WsConnection +import reactor.core.Exceptions +import spock.lang.Specification + +import java.time.Duration + +class JsonRpcWsClientSpec extends Specification { + + def "Produce error if WS is not connected"() { + setup: + def ws = Mock(WsConnection) + def client = new JsonRpcWsClient(ws) + when: + client.read(new JsonRpcRequest("foo_bar", [], 1)) + .block(Duration.ofSeconds(1)) + then: + def t = thrown(Exceptions.ReactiveException) + t.cause instanceof JsonRpcException + 1 * ws.isConnected() >> false + } +} diff --git a/src/test/resources/upstreams-ethereum-pos.yaml b/src/test/resources/upstreams-ethereum-pos.yaml index b2a1a180..4bbaf206 100644 --- a/src/test/resources/upstreams-ethereum-pos.yaml +++ b/src/test/resources/upstreams-ethereum-pos.yaml @@ -6,4 +6,4 @@ execution: rpc: url: "http://34.106.60.110:8545" - block-priority: 100 \ No newline at end of file + upstream-rating: 100 \ No newline at end of file