Pass http pool params via config (#703)

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -124,7 +124,13 @@ data class UpstreamsConfig(
         val port: Int,
     )
 
-    data class HttpEndpoint(val url: URI) {
+    data class HttpEndpoint(
+        val url: URI,
+        val maxConnections: Int,
+        val queueSize: Int,
+    ) {
+        constructor(url: URI) : this(url, DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE)
+
         var basicAuth: AuthConfig.ClientBasicAuth? = null
         var tls: AuthConfig.ClientTlsAuth? = null
     }

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -27,6 +27,9 @@ import java.io.InputStream
 import java.net.URI
 import java.util.Locale
 
+const val DEFAULT_MAX_CONNECTIONS = 1500
+const val DEFAULT_QUEUE_SIZE = 1000
+
 class UpstreamsConfigReader(
     private val fileResolver: FileResolver,
     private val optionsReader: ChainOptionsReader,
@@ -138,7 +141,7 @@ class UpstreamsConfigReader(
 
         getMapping(connConfigNode, "esplora")?.let { node ->
             getValueAsString(node, "url")?.let { url ->
-                val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                val http = UpstreamsConfig.HttpEndpoint(URI(url), DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE)
                 http.basicAuth = authConfigReader.readClientBasicAuth(node)
                 http.tls = authConfigReader.readClientTls(node)
                 connection.esplora = http
@@ -168,8 +171,11 @@ class UpstreamsConfigReader(
 
     private fun readRpcConfig(connConfigNode: MappingNode): UpstreamsConfig.HttpEndpoint? {
         return getMapping(connConfigNode, "rpc")?.let { node ->
+            val maxConnections = getValueAsInt(node, "max-connections") ?: DEFAULT_MAX_CONNECTIONS
+            val queueSize = getValueAsInt(node, "queue-size") ?: DEFAULT_QUEUE_SIZE
+
             getValueAsString(node, "url")?.let { url ->
-                val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                val http = UpstreamsConfig.HttpEndpoint(URI(url), maxConnections, queueSize)
                 http.basicAuth = authConfigReader.readClientBasicAuth(node)
                 http.tls = authConfigReader.readClientTls(node)
                 http

src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt

@@ -66,7 +66,7 @@ open class GenericConnectorFactoryCreator(
                 }
             }
             urls?.add(endpoint.url)
-            BasicHttpFactory(endpoint.url.toString(), conn.basicAuth, tls)
+            BasicHttpFactory(endpoint.url.toString(), endpoint.maxConnections, endpoint.queueSize, conn.basicAuth, tls)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt

@@ -9,13 +9,20 @@ 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 org.slf4j.LoggerFactory
 
 class BasicHttpFactory(
     private val url: String,
+    private val maxConnections: Int,
+    private val queueSize: Int,
     private val basicAuth: AuthConfig.ClientBasicAuth?,
     private val tls: ByteArray?,
 ) : HttpFactory {
+    private val log = LoggerFactory.getLogger(this::class.java)
+
     override fun create(id: String?, chain: Chain): HttpReader {
+        log.info("Creating http pool for {} with maxConnections {} and queueSize {}", url, maxConnections, queueSize)
+
         val metricsTags = listOf(
             // "unknown" is not supposed to happen
             Tag.of("upstream", id ?: "unknown"),
@@ -35,8 +42,8 @@ class BasicHttpFactory(
         )
 
         if (chain.type.apiType == ApiType.REST) {
-            return RestHttpReader(url, metrics, basicAuth, tls)
+            return RestHttpReader(url, maxConnections, queueSize, metrics, basicAuth, tls)
         }
-        return JsonRpcHttpReader(url, metrics, basicAuth, tls)
+        return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, basicAuth, tls)
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt

@@ -22,19 +22,21 @@ import java.util.function.Function
 
 abstract class HttpReader(
     protected val target: String,
+    maxConnections: Int,
+    queueSize: Int,
     protected val metrics: RequestMetrics?,
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
 ) : ChainReader {
 
-    constructor() : this("", null)
+    constructor() : this("", 1500, 1000, null)
 
     protected val httpClient: HttpClient
 
     init {
         val connectionProvider = ConnectionProvider.builder("dshackleConnectionPool")
-            .maxConnections(1500)
+            .maxConnections(maxConnections)
-            .pendingAcquireMaxCount(1000)
+            .pendingAcquireMaxCount(queueSize)
             .pendingAcquireTimeout(Duration.ofSeconds(10))
             .build()
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt

@@ -22,10 +22,12 @@ import java.util.concurrent.TimeUnit
 
 class RestHttpReader(
     target: String,
+    maxConnections: Int,
+    queueSize: Int,
     metrics: RequestMetrics,
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
-) : HttpReader(target, metrics, basicAuth, tlsCAAuth) {
+) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) {
 
     private val parser = ResponseRpcParser()
     private val requestParser = RestRequestParser

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt

@@ -37,10 +37,12 @@ import java.util.function.Function
  */
 class JsonRpcHttpReader(
     target: String,
+    maxConnections: Int,
+    queueSize: Int,
     metrics: RequestMetrics,
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
-) : HttpReader(target, metrics, basicAuth, tlsCAAuth) {
+) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) {
 
     private val parser = ResponseRpcParser()
     private val streamParser = JsonRpcStreamParser()

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy

@@ -51,7 +51,7 @@ class JsonRpcHttpReaderSpec extends Specification {
 
     def "Make a request"() {
         setup:
-        JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", metrics,null, null)
+        JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics,null, null)
         def resp = '{' +
                 '  "jsonrpc": "2.0",' +
                 '  "result": "0x98de45",' +
@@ -72,7 +72,7 @@ class JsonRpcHttpReaderSpec extends Specification {
 
     def "Produces RPC Exception on error status code"() {
         setup:
-        def client = new JsonRpcHttpReader("localhost:${port}", metrics, null, null)
+        def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, null, null)
 
         mockServer.when(
                 HttpRequest.request()
@@ -96,7 +96,7 @@ class JsonRpcHttpReaderSpec extends Specification {
 
     def "Tries to extract message if HTTP error if it still contains a JSON RPC message"() {
         setup:
-        def client = new JsonRpcHttpReader("localhost:${port}", metrics, null, null)
+        def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, null, null)
 
         mockServer.when(
                 HttpRequest.request()