diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index d71e8589..1dd94927 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -768,6 +768,9 @@ Default is 5Mb Ex `1kb`, `1024` (same as `1kb), `2mb`, etc. Default is 15Mb +| `ws.connections` +| How many concurrent connection to make. If more than one, each used in a robin-round fashion. +Defaults is `1` |=== ==== PoS Ethereum Connection Options diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 84d491d9..0249d97d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -137,6 +137,7 @@ open class UpstreamsConfig { var basicAuth: AuthConfig.ClientBasicAuth? = null var frameSize: Int? = null var msgSize: Int? = null + var connections: Int = 1 } // TODO make it unmodifiable after initial load diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 5860b990..3eaa9e60 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -126,14 +126,8 @@ class UpstreamsConfigReader( private fun readBitcoinConnection(connConfigNode: MappingNode): UpstreamsConfig.BitcoinConnection { val connection = UpstreamsConfig.BitcoinConnection() - getMapping(connConfigNode, "rpc")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) - connection.rpc = http - http.basicAuth = authConfigReader.readClientBasicAuth(node) - http.tls = authConfigReader.readClientTls(node) - } - } + .apply { rpc = readRpcConfig(connConfigNode) } + getMapping(connConfigNode, "esplora")?.let { node -> getValueAsString(node, "url")?.let { url -> val http = UpstreamsConfig.HttpEndpoint(URI(url)) @@ -164,6 +158,17 @@ class UpstreamsConfigReader( return connection } + private fun readRpcConfig(connConfigNode: MappingNode): UpstreamsConfig.HttpEndpoint? { + return getMapping(connConfigNode, "rpc")?.let { node -> + getValueAsString(node, "url")?.let { url -> + val http = UpstreamsConfig.HttpEndpoint(URI(url)) + http.basicAuth = authConfigReader.readClientBasicAuth(node) + http.tls = authConfigReader.readClientTls(node) + http + } + } + } + private fun readEthereumPosConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumPosConnection { val connection = UpstreamsConfig.EthereumPosConnection() getMapping(connConfigNode, "execution")?.let { @@ -174,16 +179,11 @@ class UpstreamsConfigReader( } return connection } + private fun readEthereumConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumConnection { val connection = UpstreamsConfig.EthereumConnection() - getMapping(connConfigNode, "rpc")?.let { node -> - getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) - connection.rpc = http - http.basicAuth = authConfigReader.readClientBasicAuth(node) - http.tls = authConfigReader.readClientTls(node) - } - } + .apply { rpc = readRpcConfig(connConfigNode) } + getValueAsBool(connConfigNode, "prefer-http")?.let { connection.preferHttp = it } @@ -208,6 +208,12 @@ class UpstreamsConfigReader( } ws.msgSize = it } + getValueAsInt(node, "connections")?.let { + if (it < 1 || it > 1024) { + throw IllegalStateException("connection limit should be in 1..1024") + } + ws.connections = it + } } } return connection diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/RawJsonBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/RawJsonBuilder.kt deleted file mode 100644 index 8b8d14c8..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/RawJsonBuilder.kt +++ /dev/null @@ -1,47 +0,0 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * Copyright (c) 2020 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.data - -import org.slf4j.LoggerFactory -import java.io.ByteArrayOutputStream - -class RawJsonBuilder { - - companion object { - private val log = LoggerFactory.getLogger(RawJsonBuilder::class.java) - - private val START = "{\"jsonrpc\":\"2.0\"".toByteArray() - private val ID_START = "\"id\":".toByteArray() - private val RESULT_START = "\"result\":".toByteArray() - private val COMMA = ",".toByteArray() - private val END = "}".toByteArray() - } - - fun write(id: Int, data: ByteArray): ByteArray { - val buf = ByteArrayOutputStream(data.size + 100) - buf.write(START) - buf.write(COMMA) - buf.write(ID_START) - buf.write(id.toString().toByteArray()) - buf.write(COMMA) - buf.write(RESULT_START) - buf.write(data) - buf.write(END) - - return buf.toByteArray() - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 5f0c3a9a..ebd86bde 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -35,7 +35,8 @@ import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumBlockValidator import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice @@ -337,17 +338,21 @@ open class ConfiguredUpstreams( chain: Chain, conn: UpstreamsConfig.EthereumConnection, urls: ArrayList? = null - ): EthereumWsFactory? { + ): EthereumWsConnectionPoolFactory? { return conn.ws?.let { endpoint -> - val wsApi = EthereumWsFactory( + val wsConnectionFactory = EthereumWsConnectionFactory( id, chain, endpoint.url, endpoint.origin ?: URI("http://localhost"), - ) - wsApi.config = endpoint - endpoint.basicAuth?.let { auth -> - wsApi.basicAuth = auth + ).apply { + config = endpoint + basicAuth = endpoint.basicAuth } + val wsApi = EthereumWsConnectionPoolFactory( + id, + endpoint.connections, + wsConnectionFactory + ) urls?.add(endpoint.url) wsApi } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt similarity index 56% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt index 55a5fee4..a2228df3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt @@ -1,25 +1,8 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * Copyright (c) 2019 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Chain 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.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics @@ -27,7 +10,7 @@ import io.micrometer.core.instrument.Tag import io.micrometer.core.instrument.Timer import java.net.URI -class EthereumWsFactory( +open class EthereumWsConnectionFactory( private val id: String, private val chain: Chain, private val uri: URI, @@ -37,15 +20,15 @@ class EthereumWsFactory( var basicAuth: AuthConfig.ClientBasicAuth? = null var config: UpstreamsConfig.WsEndpoint? = null - // metrics are shared between all connections to the same WS - private val metrics: RpcMetrics = run { + private fun metrics(connIndex: Int): RpcMetrics { val metricsTags = listOf( + Tag.of("index", connIndex.toString()), Tag.of("upstream", id), // UNSPECIFIED shouldn't happen too Tag.of("chain", chain.chainCode) ) - RpcMetrics( + return RpcMetrics( Timer.builder("upstream.ws.conn") .description("Request time through a WebSocket JSON RPC connection") .tags(metricsTags) @@ -58,11 +41,8 @@ class EthereumWsFactory( ) } - fun create(upstream: DefaultUpstream?): WsConnectionImpl { - require(upstream == null || upstream.getId() == id) { - "Creating instance for different upstream. ${upstream?.getId()} != id" - } - return WsConnectionImpl(id, uri, origin, basicAuth, metrics, upstream).also { ws -> + open fun createWsConnection(connIndex: Int = 0, onDisconnect: () -> Unit): WsConnection = + WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), onDisconnect).also { ws -> config?.frameSize?.let { ws.frameSize = it } @@ -70,5 +50,4 @@ class EthereumWsFactory( ws.msgSizeLimit = it } } - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionPoolFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionPoolFactory.kt new file mode 100644 index 00000000..3d76b6c7 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionPoolFactory.kt @@ -0,0 +1,37 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.DefaultUpstream + +class EthereumWsConnectionPoolFactory( + private val id: String, + private val connections: Int, + private val ethereumWsConnectionFactory: EthereumWsConnectionFactory +) { + + fun create(upstream: DefaultUpstream?): WsConnectionPool { + require(upstream == null || upstream.getId() == id) { + "Creating instance for different upstream. ${upstream?.getId()} != id" + } + return if (connections > 1) { + WsConnectionMultiPool(ethereumWsConnectionFactory, upstream, connections) + } else { + WsConnectionSinglePool(ethereumWsConnectionFactory, upstream) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt new file mode 100644 index 00000000..a3c51345 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2022 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +interface WsConnection : AutoCloseable { + + val isConnected: Boolean + + fun getSubscribeResponses(): Flux + fun callRpc(originalRequest: JsonRpcRequest): Mono + fun connect() +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt index 227af6f0..1168e98f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt @@ -18,8 +18,6 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.AuthConfig -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 @@ -28,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.etherjar.rpc.RpcResponseError +import io.micrometer.core.instrument.Metrics import io.netty.buffer.ByteBuf import io.netty.buffer.ByteBufInputStream import io.netty.buffer.Unpooled @@ -61,13 +60,12 @@ import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger open class WsConnectionImpl( - private val id: String, private val uri: URI, private val origin: URI, private val basicAuth: AuthConfig.ClientBasicAuth?, private val rpcMetrics: RpcMetrics?, - private val upstream: DefaultUpstream?, -) : AutoCloseable { + private val onDisconnect: () -> Unit +) : AutoCloseable, WsConnection, Cloneable { companion object { private val log = LoggerFactory.getLogger(WsConnectionImpl::class.java) @@ -124,7 +122,7 @@ open class WsConnectionImpl( private var connection: Disposable? = null private val reconnecting = AtomicBoolean(false) - open val isConnected: Boolean + override val isConnected: Boolean get() = connection != null && !reconnecting.get() fun setReconnectIntervalSeconds(value: Long) { @@ -132,7 +130,7 @@ open class WsConnectionImpl( currentBackOff = reconnectBackoff.start() } - fun connect() { + override fun connect() { keepConnection = true connectInternal() } @@ -179,10 +177,9 @@ open class WsConnectionImpl( connection = HttpClient.create() .resolver(DefaultAddressResolverGroup.INSTANCE) .doOnDisconnected { + onDisconnect() disconnects.tryEmitNext(Instant.now()) log.info("Disconnected from $uri") - // mark upstream as UNAVAIL - upstream?.setStatus(UpstreamAvailability.UNAVAILABLE) if (keepConnection) { tryReconnectLater() } @@ -274,7 +271,7 @@ open class WsConnectionImpl( ) } - fun onMessage(msg: ResponseWSParser.WsResponse): Mono { + private fun onMessage(msg: ResponseWSParser.WsResponse): Mono { return Mono.fromCallable { when (msg.type) { ResponseWSParser.Type.RPC -> onMessageRpc(msg) @@ -283,7 +280,7 @@ open class WsConnectionImpl( }.then() } - fun onMessageRpc(msg: ResponseWSParser.WsResponse) { + private fun onMessageRpc(msg: ResponseWSParser.WsResponse) { val rpcResponse = JsonRpcResponse( msg.value, msg.error, msg.id, null ) @@ -302,7 +299,7 @@ open class WsConnectionImpl( } } - fun onMessageSubscription(msg: ResponseWSParser.WsResponse) { + private fun onMessageSubscription(msg: ResponseWSParser.WsResponse) { val subscription = JsonRpcWsMessage( msg.value, msg.error, msg.id.asString(), ) @@ -316,11 +313,11 @@ open class WsConnectionImpl( } } - open fun getSubscribeResponses(): Flux { + override fun getSubscribeResponses(): Flux { return Flux.from(subscriptionResponses.asFlux()) } - open fun callRpc(originalRequest: JsonRpcRequest): Mono { + override fun callRpc(originalRequest: JsonRpcRequest): Mono { return Mono.fromCallable { val startTime = System.nanoTime() // use an internal id sequence, to avoid id conflicts with user calls @@ -342,7 +339,7 @@ open class WsConnectionImpl( } } - fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono { + private fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono { val internalId = request.id.toLong() val onResponse = Sinks.one() currentRequests[internalId.toInt()] = onResponse @@ -386,5 +383,14 @@ open class WsConnectionImpl( connection?.dispose() connection = null currentRequests.clear() + rpcMetrics?.fails?.let { + it.close() + Metrics.globalRegistry.remove(it) + } + rpcMetrics?.timer?.let { + it.close() + Metrics.globalRegistry.remove(it) + } + resetBackoffExecutor.shutdown() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPool.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPool.kt new file mode 100644 index 00000000..0c886fce --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPool.kt @@ -0,0 +1,133 @@ +/** + * Copyright (c) 2022 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import org.springframework.util.backoff.BackOffExecution +import org.springframework.util.backoff.ExponentialBackOff +import java.time.Duration +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.locks.ReentrantReadWriteLock +import kotlin.concurrent.read +import kotlin.concurrent.write + +/** + * A Websocket connection pool which keeps up to `target` connection providing them in a round-robin fashion. + * + * It doesn't make all the connections immediately but rather grows them one by one ensuring existing are ok. + * By default, it adds a new connection every 5 second until it reaches the target number. + */ +class WsConnectionMultiPool( + private val ethereumWsConnectionFactory: EthereumWsConnectionFactory, + private val upstream: DefaultUpstream?, + private val connections: Int, +) : WsConnectionPool { + + companion object { + private const val SCHEDULE_FULL = 60L + private const val SCHEDULE_GROW = 5L + private const val SCHEDULE_BROKEN = 15L + } + + private val current = ArrayList() + private var adjustLock = ReentrantReadWriteLock() + private val index = AtomicInteger(0) + private var connIndex = 0 + + var scheduler: ScheduledExecutorService = Global.control + + override fun connect() { + adjust() + } + + override fun getConnection(): WsConnection { + val tries = ExponentialBackOff(50, 1.25).also { + it.maxElapsedTime = Duration.ofMinutes(1).toMillis() + }.start() + var next = next() + while (next == null) { + val sleep = tries.nextBackOff() + if (sleep == BackOffExecution.STOP) { + throw IllegalStateException("No available WS connection") + } + Thread.sleep(sleep) + next = next() + } + return next + } + + override fun close() { + adjustLock.write { + current.forEach { it.close() } + current.clear() + } + } + + private fun next(): WsConnection? { + adjustLock.read { + if (current.isEmpty()) { + return null + } + return current[index.getAndIncrement() % current.size] + } + } + + private fun adjust() { + adjustLock.write { + // add a new connection only if all existing are active or there are no connections at all + val allOk = current.all { it.isConnected } + val schedule: Long + if (allOk) { + schedule = if (current.size >= connections) { + // recheck the state in a minute and adjust if any connection went bad + SCHEDULE_FULL + } else { + current.add( + ethereumWsConnectionFactory.createWsConnection(connIndex++) { + if (isUnavailable()) { + upstream?.setStatus(UpstreamAvailability.UNAVAILABLE) + } + }.also { + it.connect() + } + ) + SCHEDULE_GROW + } + } else { + // technically there is no reason to disconnect because it supposed to reconnect, + // but to ensure clean start (or other internal state issues) lets completely close all broken and create new + current.removeIf { + if (!it.isConnected) { + // DO NOT FORGET to close the connection, otherwise it would keep reconnecting but unused + it.close() + true + } else { + false + } + } + schedule = SCHEDULE_BROKEN + } + + scheduler.schedule({ adjust() }, schedule, TimeUnit.SECONDS) + } + } + + private fun isUnavailable() = adjustLock.read { current.count { it.isConnected } == 0 } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionPool.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionPool.kt new file mode 100644 index 00000000..ff8f17b7 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionPool.kt @@ -0,0 +1,21 @@ +/** + * Copyright (c) 2022 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +interface WsConnectionPool : AutoCloseable { + fun connect() + fun getConnection(): WsConnection +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSinglePool.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSinglePool.kt new file mode 100644 index 00000000..4540220f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSinglePool.kt @@ -0,0 +1,42 @@ +/** + * Copyright (c) 2022 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability + +class WsConnectionSinglePool( + ethereumWsConnectionFactory: EthereumWsConnectionFactory, + private val upstream: DefaultUpstream?, +) : WsConnectionPool { + private val connection = ethereumWsConnectionFactory.createWsConnection { + upstream?.setStatus(UpstreamAvailability.UNAVAILABLE) + } + + override fun connect() { + if (!connection.isConnected) { + connection.connect() + } + } + + override fun getConnection(): WsConnection { + return connection + } + + override fun close() { + connection.close() + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt index ea2c44b1..59a78a86 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt @@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicReference class WsSubscriptionsImpl( - val conn: WsConnectionImpl, + val wsPool: WsConnectionPool, ) : WsSubscriptions { companion object { @@ -35,6 +35,7 @@ class WsSubscriptionsImpl( override fun subscribe(method: String): Flux { val subscriptionId = AtomicReference("") + val conn = wsPool.getConnection() val messages = conn.getSubscribeResponses() .filter { it.subscriptionId == subscriptionId.get() } .filter { it.result != null } // should never happen diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt index 34c5c600..c307bde5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt @@ -5,13 +5,13 @@ import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.HttpFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator -import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory open class EthereumConnectorFactory( private val preferHttp: Boolean, - private val wsFactory: EthereumWsFactory?, + private val wsFactory: EthereumWsConnectionPoolFactory?, private val httpFactory: HttpFactory?, private val forkChoice: ForkChoice, private val blockValidator: BlockValidator 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 e750b884..c29d5b50 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 @@ -7,7 +7,13 @@ import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.MergedHead -import io.emeraldpay.dshackle.upstream.ethereum.* +import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription +import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead +import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumIngressSubscription +import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool +import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory @@ -15,12 +21,12 @@ import java.time.Duration class EthereumRpcConnector( private val directReader: JsonRpcReader, - wsFactory: EthereumWsFactory?, + wsFactory: EthereumWsConnectionPoolFactory?, id: String, forkChoice: ForkChoice, blockValidator: BlockValidator ) : EthereumConnector, CachesEnabled { - private val conn: WsConnectionImpl? + private val pool: WsConnectionPool? private val head: Head companion object { @@ -30,14 +36,14 @@ 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) - val subscriptions = WsSubscriptionsImpl(conn) + pool = wsFactory.create(null) + val subscriptions = WsSubscriptionsImpl(pool) val wsHead = EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), subscriptions) // receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed val rpcHead = EthereumRpcHead(getIngressReader(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30)) head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id") } else { - conn = null + pool = null log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") head = EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator) } @@ -50,7 +56,7 @@ class EthereumRpcConnector( } override fun start() { - conn?.connect() + pool?.connect() if (head is Lifecycle) { head.start() } @@ -67,7 +73,7 @@ class EthereumRpcConnector( if (head is Lifecycle) { head.stop() } - conn?.close() + pool?.close() } override fun getIngressReader(): JsonRpcReader { 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 f7127559..c965cc79 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 @@ -4,32 +4,36 @@ import io.emeraldpay.dshackle.reader.JsonRpcReader import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.ethereum.* +import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead +import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool +import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient class EthereumWsConnector( - wsFactory: EthereumWsFactory, + wsFactory: EthereumWsConnectionPoolFactory, upstream: DefaultUpstream, forkChoice: ForkChoice, blockValidator: BlockValidator ) : EthereumConnector { - private val conn: WsConnectionImpl + private val pool: WsConnectionPool private val reader: JsonRpcReader private val head: EthereumWsHead private val subscriptions: EthereumIngressSubscription init { - conn = wsFactory.create(upstream) - reader = JsonRpcWsClient(conn) - val wsSubscriptions = WsSubscriptionsImpl(conn) + pool = wsFactory.create(upstream) + reader = JsonRpcWsClient(pool) + val wsSubscriptions = WsSubscriptionsImpl(pool) head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, reader, wsSubscriptions) subscriptions = EthereumWsIngressSubscription(wsSubscriptions) } override fun start() { - conn.connect() + pool.connect() head.start() } @@ -38,7 +42,7 @@ class EthereumWsConnector( } override fun stop() { - conn.close() + pool.close() head.stop() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt deleted file mode 100644 index c9b57de5..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt +++ /dev/null @@ -1,27 +0,0 @@ -package io.emeraldpay.dshackle.upstream.rpcclient - -import io.emeraldpay.dshackle.reader.JsonRpcReader -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: JsonRpcReader, - private val secondary: JsonRpcReader, -) : JsonRpcReader { - - 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 bd60dd49..23948c53 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt @@ -16,16 +16,17 @@ package io.emeraldpay.dshackle.upstream.rpcclient import io.emeraldpay.dshackle.reader.JsonRpcReader -import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl +import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool import io.emeraldpay.etherjar.rpc.RpcResponseError import reactor.core.publisher.Mono class JsonRpcWsClient( - private val ws: WsConnectionImpl + private val wsPool: WsConnectionPool ) : JsonRpcReader { override fun read(key: JsonRpcRequest): Mono { - if (!ws.isConnected) { + val conn = wsPool.getConnection() + if (!conn.isConnected) { return Mono.error( JsonRpcException( JsonRpcResponse.NumberId(key.id), @@ -36,6 +37,6 @@ class JsonRpcWsClient( ) ) } - return ws.callRpc(key) + return conn.callRpc(key) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy index 86559520..761402b7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy @@ -19,7 +19,7 @@ class WsConnectionImplRealSpec extends Specification { @Shared MockWSServer server @Shared - WsConnectionImpl conn + WsConnection conn def setup() { if (System.getenv("CI") == "true") { @@ -31,7 +31,16 @@ class WsConnectionImplRealSpec extends Specification { server = new MockWSServer(port) server.start() Thread.sleep(SLEEP) - conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null) + conn = new EthereumWsConnectionPoolFactory( + "test", + 1, + new EthereumWsConnectionFactory( + "test", + Chain.ETHEREUM, + "ws://localhost:${port}".toURI(), + "http://localhost:${port}".toURI() + ) + ).create(null).getConnection() } def cleanup() { @@ -101,7 +110,16 @@ class WsConnectionImplRealSpec extends Specification { def up = Mock(DefaultUpstream) { _ * getId() >> "test" } - conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up) + conn = new EthereumWsConnectionPoolFactory( + "test", + 1, + new EthereumWsConnectionFactory( + "test", + Chain.ETHEREUM, + "ws://localhost:${port}".toURI(), + "http://localhost:${port}".toURI() + ) + ).create(up).getConnection() when: conn.connect() conn.reconnectIntervalSeconds = 10 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy index a8017242..c8a56dfc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy @@ -15,33 +15,36 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId 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.dshackle.Chain import reactor.core.publisher.Flux import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration -import java.time.Instant -import java.time.temporal.ChronoUnit class WsConnectionImplSpec extends Specification { def "Makes a RPC call"() { setup: - def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsConnectionPoolFactory( + "test", + 1, + new EthereumWsConnectionFactory( + "test", + Chain.ETHEREUM, + new URI("http://localhost"), + new URI("http://localhost") + ) + ) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null) + def ws = wsf.create(null).getConnection() as WsConnectionImpl def tx = new TransactionJson().tap { hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") @@ -63,10 +66,19 @@ class WsConnectionImplSpec extends Specification { def "Makes a RPC call - return null"() { setup: - def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsConnectionPoolFactory( + "test", + 1, + new EthereumWsConnectionFactory( + "test", + Chain.ETHEREUM, + new URI("http://localhost"), + new URI("http://localhost") + ) + ) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null) + def ws = wsf.create(null).getConnection() apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null) @@ -86,10 +98,19 @@ class WsConnectionImplSpec extends Specification { def "Makes a RPC call - return error"() { setup: - def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost")) + def wsf = new EthereumWsConnectionPoolFactory( + "test", + 1, + new EthereumWsConnectionFactory( + "test", + Chain.ETHEREUM, + new URI("http://localhost"), + new URI("http://localhost") + ) + ) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create(null) + def ws = wsf.create(null).getConnection() apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test")) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPoolSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPoolSpec.groovy new file mode 100644 index 00000000..03929468 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPoolSpec.groovy @@ -0,0 +1,140 @@ +/** + * Copyright (c) 2022 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import spock.lang.Specification + +import java.util.concurrent.ScheduledExecutorService + +class WsConnectionMultiPoolSpec extends Specification { + + def "create connection when less than required"() { + setup: + def conn = Mock(WsConnection) + def up = Mock(DefaultUpstream) + def factory = Mock(EthereumWsConnectionFactory) + def pool = new WsConnectionMultiPool(factory, up, 3) + pool.scheduler = Stub(ScheduledExecutorService) + + when: + pool.connect() + + then: + 1 * factory.createWsConnection(0, _) >> conn + 1 * conn.connect() + } + + def "create connection until target"() { + setup: + def conn1 = Mock(WsConnection) + def conn2 = Mock(WsConnection) + def conn3 = Mock(WsConnection) + def up = Mock(DefaultUpstream) + def factory = Mock(EthereumWsConnectionFactory) + def pool = new WsConnectionMultiPool(factory, up, 3) + pool.scheduler = Stub(ScheduledExecutorService) + + when: + pool.connect() + + then: + 1 * factory.createWsConnection(0, _) >> conn1 + 1 * conn1.connect() + + when: + pool.connect() + + then: + 1 * conn1.isConnected() >> true + 1 * factory.createWsConnection(1, _) >> conn2 + 1 * conn2.connect() + + when: + pool.connect() + + then: + 1 * conn1.isConnected() >> true + 1 * conn2.isConnected() >> true + 1 * factory.createWsConnection(2, _) >> conn3 + 1 * conn3.connect() + + when: + pool.connect() + + then: + 1 * conn1.isConnected() >> true + 1 * conn2.isConnected() >> true + 1 * conn3.isConnected() >> true + 0 * factory.createWsConnection(_, _) + } + + def "recreate connection after failure"() { + setup: + def conn1 = Mock(WsConnection) + def conn2 = Mock(WsConnection) + def conn3 = Mock(WsConnection) + def conn4 = Mock(WsConnection) + def up = Mock(DefaultUpstream) + def factory = Mock(EthereumWsConnectionFactory) + def pool = new WsConnectionMultiPool(factory, up, 3) + pool.scheduler = Stub(ScheduledExecutorService) + + when: "initial fill" + pool.connect() + pool.connect() + pool.connect() + + then: + _ * conn1.isConnected() >> true + _ * conn2.isConnected() >> true + _ * conn3.isConnected() >> true + 1 * factory.createWsConnection(0, _) >> conn1 + 1 * factory.createWsConnection(1, _) >> conn2 + 1 * factory.createWsConnection(2, _) >> conn3 + 1 * conn1.connect() + 1 * conn2.connect() + 1 * conn3.connect() + + when: "all ok" + pool.connect() + + then: + 1 * conn1.isConnected() >> true + 1 * conn2.isConnected() >> true + 1 * conn3.isConnected() >> true + 0 * factory.createWsConnection(_, _) + + when: "one failed" + pool.connect() + + then: + (1.._) * conn1.isConnected() >> true + (1.._) * conn2.isConnected() >> false + (1.._) * conn3.isConnected() >> true + 0 * factory.createWsConnection(_, _) // doesn't create immediately, but schedules it for the next adjust + 1 * conn2.close() + + when: "needs one more" + pool.connect() + + then: + 1 * conn1.isConnected() >> true + 1 * conn3.isConnected() >> true + 1 * factory.createWsConnection(3, _) >> conn4 + 1 * conn4.connect() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy index 7a096106..9491879a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImplSpec.groovy @@ -20,8 +20,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.publisher.Sinks -import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration @@ -38,8 +36,11 @@ class WsSubscriptionsImplSpec extends Specification { ] ) - def conn = Mock(WsConnectionImpl) - def ws = new WsSubscriptionsImpl(conn) + def conn = Mock(WsConnection) + def pool = Mock(WsConnectionPool) { + getConnection() >> conn + } + def ws = new WsSubscriptionsImpl(pool) when: def act = ws.subscribe("foo_bar") @@ -70,8 +71,11 @@ class WsSubscriptionsImplSpec extends Specification { ] ) - def conn = Mock(WsConnectionImpl) - def ws = new WsSubscriptionsImpl(conn) + def conn = Mock(WsConnection) + def pool = Mock(WsConnectionPool) { + getConnection() >> conn + } + def ws = new WsSubscriptionsImpl(pool) when: def act = ws.subscribe("foo_bar") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy deleted file mode 100644 index 883e400b..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy +++ /dev/null @@ -1,70 +0,0 @@ -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 index 77b30435..f5844b72 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy @@ -1,6 +1,7 @@ package io.emeraldpay.dshackle.upstream.rpcclient -import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl +import io.emeraldpay.dshackle.upstream.ethereum.WsConnection +import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool import reactor.core.Exceptions import spock.lang.Specification @@ -10,8 +11,11 @@ class JsonRpcWsClientSpec extends Specification { def "Produce error if WS is not connected"() { setup: - def ws = Mock(WsConnectionImpl) - def client = new JsonRpcWsClient(ws) + def ws = Mock(WsConnection) + def pool = Mock(WsConnectionPool) { + getConnection() >> ws + } + def client = new JsonRpcWsClient(pool) when: client.read(new JsonRpcRequest("foo_bar", [], 1)) .block(Duration.ofSeconds(1))