Move ws connection to separate scheduler

drpcorg · Oct 30, 2023 · 93655a2 · 93655a2
1 parent 4ea4fa5
commit 93655a2
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 2 deletions.
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt
@@ -30,6 +30,11 @@ open class SchedulersConfig {
         return makeScheduler("head-scheduler", 4, monitoringConfig)
     }
 
+    @Bean
+    open fun wsScheduler(monitoringConfig: MonitoringConfig): Scheduler {
+        return makeScheduler("ws-scheduler", 4, monitoringConfig)
+    }
+
     @Bean
     open fun wsConnectionResubscribeScheduler(monitoringConfig: MonitoringConfig): Scheduler {
         return makeScheduler("ws-connection-resubscribe-scheduler", 2, monitoringConfig)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -98,6 +98,8 @@ open class ConfiguredUpstreams(
     private val clientSpansInterceptor: ClientInterceptor?,
     @Qualifier("headScheduler")
     private val headScheduler: Scheduler,
+    @Qualifier("wsScheduler")
+    private val wsScheduler: Scheduler,
     private val authorizationConfig: AuthorizationConfig,
     private val grpcAuthContext: GrpcAuthContext,
 ) : ApplicationRunner {
@@ -482,7 +484,7 @@ open class ConfiguredUpstreams(
                 chain,
                 endpoint.url,
                 endpoint.origin ?: URI("http://localhost"),
-                headScheduler,
+                wsScheduler,
             ).apply {
                 config = endpoint
                 basicAuth = endpoint.basicAuth

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt
@@ -339,7 +339,7 @@ open class WsConnectionImpl(
         val status = subscriptionResponses.tryEmitNext(subscription)
         if (status.isFailure) {
             if (status == Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) {
-                log.debug("No subscribers to WS response")
+                log.debug("No subscribers to WS response with id {} and result {}", msg.id, msg.value?.decodeToString())
             } else {
                 log.warn("Failed to proceed with a WS message: $status")
             }