Remove eventPublisher and locks (#336)

src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt

@@ -25,14 +25,15 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
         @Qualifier("headScheduler")
         headScheduler: Scheduler,
         tracer: Tracer,
+        multistreamEventsScheduler: Scheduler,
     ): List<Multistream> {
         return Chain.entries
             .filterNot { it == Chain.UNSPECIFIED }
             .map { chain ->
                 if (chain.type == BITCOIN) {
-                    bitcoinMultistream(chain, cachesFactory, headScheduler)
+                    bitcoinMultistream(chain, cachesFactory, headScheduler, multistreamEventsScheduler)
                 } else {
-                    genericMultistream(chain, cachesFactory, headScheduler, tracer)
+                    genericMultistream(chain, cachesFactory, headScheduler, tracer, multistreamEventsScheduler)
                 }
             }
     }
@@ -42,13 +43,17 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
         cachesFactory: CachesFactory,
         headScheduler: Scheduler,
         tracer: Tracer,
+        multistreamEventsScheduler: Scheduler,
     ): Multistream {
         val name = "multi-$chain"
         val cs = ChainSpecificRegistry.resolve(chain)
+        val caches = cachesFactory.getCaches(chain)
         return GenericMultistream(
             chain,
+            multistreamEventsScheduler,
+            cs.callSelector(caches),
             CopyOnWriteArrayList(),
-            cachesFactory.getCaches(chain),
+            caches,
             headScheduler,
             cs.makeCachingReaderBuilder(tracer),
             cs::localReaderBuilder,
@@ -60,11 +65,13 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
         chain: Chain,
         cachesFactory: CachesFactory,
         headScheduler: Scheduler,
+        multistreamEventsScheduler: Scheduler,
     ): BitcoinMultistream {
         val name = "multi-bitcoin-$chain"
 
         return BitcoinMultistream(
             chain,
+            multistreamEventsScheduler,
             ArrayList(),
             cachesFactory.getCaches(chain),
             headScheduler,

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 multistreamEventsScheduler(monitoringConfig: MonitoringConfig): Scheduler {
+        return makeScheduler("events-scheduler", 4, monitoringConfig)
+    }
+
     @Bean
     open fun wsConnectionResubscribeScheduler(monitoringConfig: MonitoringConfig): Scheduler {
         return makeScheduler("ws-connection-resubscribe-scheduler", 2, monitoringConfig)

src/main/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigUpstreamService.kt

@@ -6,12 +6,10 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
 import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
-import org.springframework.context.ApplicationEventPublisher
 import org.springframework.stereotype.Component
 
 @Component
 open class ReloadConfigUpstreamService(
-    private val eventPublisher: ApplicationEventPublisher,
     private val multistreamHolder: CurrentMultistreamHolder,
     private val configuredUpstreams: ConfiguredUpstreams,
 ) {
@@ -44,19 +42,19 @@ open class ReloadConfigUpstreamService(
 
         chainsToReload.forEach {
             usedChains.add(it)
-            multistreamHolder.getUpstream(it)
+            val ms = multistreamHolder.getUpstream(it)
-                .getAll()
+            ms.getAll()
                 .forEach { up ->
-                    eventPublisher.publishEvent(UpstreamChangeEvent(it, up, UpstreamChangeEvent.ChangeType.REMOVED))
+                    ms.processUpstreamsEvents(UpstreamChangeEvent(it, up, UpstreamChangeEvent.ChangeType.REMOVED))
                 }
         }
         upstreamsToRemove.forEach { pair ->
             usedChains.add(pair.second)
-            multistreamHolder.getUpstream(pair.second)
+            val ms = multistreamHolder.getUpstream(pair.second)
-                .getAll()
+            ms.getAll()
                 .find { pair.first == it.getId() }
-                ?.let {
+                ?.let { up ->
-                    eventPublisher.publishEvent(UpstreamChangeEvent(pair.second, it, UpstreamChangeEvent.ChangeType.REMOVED))
+                    ms.processUpstreamsEvents(UpstreamChangeEvent(pair.second, up, UpstreamChangeEvent.ChangeType.REMOVED))
                 }
         }
 

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -19,7 +19,6 @@ package io.emeraldpay.dshackle.rpc
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
-import io.emeraldpay.dshackle.BlockchainType.ETHEREUM
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.Global.Companion.nullValue
@@ -37,13 +36,11 @@ import io.emeraldpay.dshackle.reader.RpcReader
 import io.emeraldpay.dshackle.reader.RpcReaderFactory
 import io.emeraldpay.dshackle.reader.RpcReaderFactory.RpcReaderData
 import io.emeraldpay.dshackle.reader.SpannedReader
-import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.ApiSource
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
-import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -56,13 +53,11 @@ import org.slf4j.LoggerFactory
 import org.springframework.cloud.sleuth.Span
 import org.springframework.cloud.sleuth.Tracer
 import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth
-import org.springframework.context.event.EventListener
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.kotlin.core.publisher.toMono
 import reactor.util.context.Context
-import java.util.EnumMap
 
 @Service
 open class NativeCall(
@@ -78,19 +73,6 @@ open class NativeCall(
     private val passthrough = config.passthrough
 
     var rpcReaderFactory: RpcReaderFactory = RpcReaderFactory.default()
-    private val ethereumCallSelectors = EnumMap<Chain, EthereumCallSelector>(Chain::class.java)
-
-    @EventListener
-    fun onUpstreamChangeEvent(event: UpstreamChangeEvent) {
-        multistreamHolder.getUpstream(event.chain).let { up ->
-            if (up.chain.type == ETHEREUM) {
-                ethereumCallSelectors.putIfAbsent(
-                    event.chain,
-                    EthereumCallSelector(up.caches),
-                )
-            }
-        }
-    }
 
     open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
         return nativeCallResult(requestMono)
@@ -305,11 +287,7 @@ open class NativeCall(
         }
         // for ethereum the actual block needed for the call may be specified in the call parameters
         val callSpecificMatcher: Mono<Selector.Matcher> =
-            if (upstream.chain.type == ETHEREUM) {
+            upstream.callSelector?.getMatcher(method, params, upstream.getHead(), passthrough) ?: Mono.empty()
-                ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead(), passthrough)
-            } else {
-                null
-            } ?: Mono.empty()
         return callSpecificMatcher.defaultIfEmpty(Selector.empty).map { csm ->
             val matcher = Selector.Builder()
                 .withMatcher(csm)

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -16,97 +16,24 @@
  */
 package io.emeraldpay.dshackle.startup
 
-import brave.grpc.GrpcTracing
-import com.google.common.annotations.VisibleForTesting
-import io.emeraldpay.dshackle.BlockchainType.BITCOIN
-import io.emeraldpay.dshackle.BlockchainType.ETHEREUM
 import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.Defaults
-import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.Global
-import io.emeraldpay.dshackle.config.AuthorizationConfig
-import io.emeraldpay.dshackle.config.ChainsConfig
-import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
-import io.emeraldpay.dshackle.config.CompressionConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.config.UpstreamsConfig.BitcoinConnection
-import io.emeraldpay.dshackle.config.UpstreamsConfig.EthereumPosConnection
-import io.emeraldpay.dshackle.config.UpstreamsConfig.HttpEndpoint
-import io.emeraldpay.dshackle.config.UpstreamsConfig.RpcConnection
 import io.emeraldpay.dshackle.foundation.ChainOptions
-import io.emeraldpay.dshackle.foundation.ChainOptions.Options
+import io.emeraldpay.dshackle.startup.configure.UpstreamFactory
-import io.emeraldpay.dshackle.upstream.BlockValidator
+import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
-import io.emeraldpay.dshackle.upstream.CallTargetsHolder
-import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.HttpRpcFactory
-import io.emeraldpay.dshackle.upstream.MergedHead
-import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead
-import io.emeraldpay.dshackle.upstream.bitcoin.EsploraClient
-import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
-import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer
-import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionFactory
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
-import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
-import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
-import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
-import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
-import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory
-import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
-import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
-import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext
-import io.grpc.ClientInterceptor
 import org.slf4j.LoggerFactory
-import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.beans.factory.annotation.Qualifier
-import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.ApplicationArguments
 import org.springframework.boot.ApplicationRunner
-import org.springframework.context.ApplicationEventPublisher
 import org.springframework.stereotype.Component
-import reactor.core.scheduler.Scheduler
-import reactor.core.scheduler.Schedulers
-import java.net.URI
-import java.util.concurrent.Executor
-import java.util.concurrent.Executors
-import java.util.concurrent.atomic.AtomicInteger
-import java.util.function.Function
-import kotlin.math.abs
 
 @Component
 open class ConfiguredUpstreams(
-    private val fileResolver: FileResolver,
+    private val upstreamFactory: UpstreamFactory,
     private val config: UpstreamsConfig,
-    private val compressionConfig: CompressionConfig,
+    private val multistreamHolder: CurrentMultistreamHolder,
-    private val callTargets: CallTargetsHolder,
-    private val eventPublisher: ApplicationEventPublisher,
-    @Qualifier("grpcChannelExecutor")
-    private val channelExecutor: Executor,
-    private val chainsConfig: ChainsConfig,
-    private val grpcTracing: GrpcTracing,
-    private val wsConnectionResubscribeScheduler: Scheduler,
-    @Autowired(required = false)
-    private val clientSpansInterceptor: ClientInterceptor?,
-    @Qualifier("headScheduler")
-    private val headScheduler: Scheduler,
-    private val wsScheduler: Scheduler,
-    private val headLivenessScheduler: Scheduler,
-    private val authorizationConfig: AuthorizationConfig,
-    private val grpcAuthContext: GrpcAuthContext,
 ) : ApplicationRunner {
-    @Value("\${spring.application.max-metadata-size}")
-    private var maxMetadataSize: Int = Defaults.maxMetadataSize
-
     private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
-    private var seq = AtomicInteger(0)
-    private val hashes: MutableMap<Byte, Boolean> = HashMap()
-
-    lateinit var grpcUpstreamsScheduler: Scheduler
 
     override fun run(args: ApplicationArguments) {
         log.debug("Starting upstreams")
@@ -121,59 +48,18 @@ open class ConfiguredUpstreams(
                 return@forEach
             }
             log.debug("Start upstream ${up.id}")
-            if (up.connection is UpstreamsConfig.GrpcConnection) {
+            if (up.connection !is UpstreamsConfig.GrpcConnection) {
-                val options = up.options ?: ChainOptions.PartialOptions()
-                buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options.buildOptions(), compressionConfig.grpc.clientEnabled)
-            } else {
                 val chain = Global.chainById(up.chain)
                 if (chain == Chain.UNSPECIFIED) {
                     log.error("Chain is unknown: ${up.chain}")
                     return@forEach
                 }
-                val chainConfig = chainsConfig.resolve(up.chain!!)
+                val upstream = upstreamFactory.createUpstream(chain.type, up, defaultOptions)
-                val options = chainConfig.options
+                if (upstream != null) {
-                    .merge(defaultOptions[chain] ?: ChainOptions.PartialOptions.getDefaults())
+                    multistreamHolder.getUpstream(chain)
-                    .merge(up.options ?: ChainOptions.PartialOptions())
+                        .processUpstreamsEvents(
-                    .buildOptions()
+                            UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED),
-                val upstream = when (chain.type) {
-                    BITCOIN -> {
-                        buildBitcoinUpstream(
-                            up.cast(BitcoinConnection::class.java),
-                            chain,
-                            options,
-                            chainConfig,
                         )
-                    }
-
-                    ETHEREUM -> {
-                        val posConn = up.cast(EthereumPosConnection::class.java)
-
-                        buildGenericUpstream(
-                            up.nodeId,
-                            up,
-                            posConn.connection?.execution ?: throw IllegalStateException("Empty execution config"),
-                            chain,
-                            options,
-                            chainConfig,
-                            posConn.connection?.upstreamRating ?: 0,
-                        )
-                    }
-
-                    else -> {
-                        buildGenericUpstream(
-                            up.nodeId,
-                            up,
-                            up.connection as RpcConnection,
-                            chain,
-                            options,
-                            chainConfig,
-                            0,
-                        )
-                    }
-                }
-                upstream?.let {
-                    val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED)
-                    eventPublisher.publishEvent(event)
                 }
             }
         }
@@ -196,264 +82,4 @@ open class ConfiguredUpstreams(
         }
         return defaultOptions
     }
-
-    fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
-        return if (config.methods != null || config.methodGroups != null) {
-            ManagedCallMethods(
-                delegate = callTargets.getDefaultMethods(chain),
-                enabled = config.methods?.enabled?.map { it.name }?.toSet() ?: emptySet(),
-                disabled = config.methods?.disabled?.map { it.name }?.toSet() ?: emptySet(),
-                groupsEnabled = config.methodGroups?.enabled ?: emptySet(),
-                groupsDisabled = config.methodGroups?.disabled ?: emptySet(),
-            ).also {
-                config.methods?.enabled?.forEach { m ->
-                    if (m.quorum != null) {
-                        it.setQuorum(m.name, m.quorum)
-                    }
-                    if (m.static != null) {
-                        it.setStaticResponse(m.name, m.static)
-                    }
-                }
-            }
-        } else {
-            callTargets.getDefaultMethods(chain)
-        }
-    }
-
-    private fun buildGenericUpstream(
-        nodeId: Int?,
-        config: UpstreamsConfig.Upstream<*>,
-        connection: RpcConnection,
-        chain: Chain,
-        options: Options,
-        chainConfig: ChainConfig,
-        nodeRating: Int,
-    ): Upstream? {
-        if (config.connection == null) {
-            log.warn("Upstream doesn't have connection configuration")
-            return null
-        }
-
-        val cs = ChainSpecificRegistry.resolve(chain)
-
-        val connectorFactory = buildConnectorFactory(
-            config.id!!,
-            connection,
-            chain,
-            NoChoiceWithPriorityForkChoice(nodeRating, config.id!!),
-            BlockValidator.ALWAYS_VALID,
-            chainConfig,
-        ) ?: return null
-
-        val methods = buildMethods(config, chain)
-
-        val hashUrl = connection.let {
-            if (it.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
-        }
-        val hash = getHash(nodeId, hashUrl!!)
-
-        val upstream = GenericUpstream(
-            config.id!!,
-            chain,
-            hash,
-            options,
-            config.role,
-            methods,
-            QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)),
-            chainConfig,
-            connectorFactory,
-            eventPublisher,
-            cs::validator,
-            cs::labelDetector,
-            cs::subscriptionTopics,
-        )
-
-        upstream.start()
-        if (!upstream.isRunning) {
-            log.debug("Upstream ${upstream.getId()} is not running, it can't be added")
-            return null
-        }
-        return upstream
-    }
-
-    private fun buildBitcoinUpstream(
-        config: UpstreamsConfig.Upstream<BitcoinConnection>,
-        chain: Chain,
-        options: Options,
-        chainConf: ChainConfig,
-    ): Upstream? {
-        val conn = config.connection!!
-        val httpFactory = buildHttpFactory(conn.rpc)
-        if (httpFactory == null) {
-            log.warn("Upstream doesn't have API configuration")
-            return null
-        }
-        val directApi = httpFactory.create(config.id, chain)
-        val esplora = conn.esplora?.let { endpoint ->
-            val tls = endpoint.tls?.let { tls ->
-                tls.ca?.let { ca ->
-                    fileResolver.resolve(ca).readBytes()
-                }
-            }
-            EsploraClient(endpoint.url, endpoint.basicAuth, tls)
-        }
-
-        val extractBlock = ExtractBlock()
-        val rpcHead = BitcoinRpcHead(directApi, extractBlock, headScheduler = headScheduler)
-        val head: Head = conn.zeroMq?.let { zeroMq ->
-            val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock")
-            val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock, headScheduler)
-            MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice(), headScheduler)
-        } ?: rpcHead
-
-        val methods = buildMethods(config, chain)
-        val upstream = BitcoinRpcUpstream(
-            config.id
-                ?: "bitcoin-${seq.getAndIncrement()}",
-            chain, directApi, head,
-            options, config.role,
-            QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)),
-            methods, esplora, chainConf,
-        )
-        upstream.start()
-        return upstream
-    }
-
-    private fun buildGrpcUpstream(
-        nodeId: Int?,
-        config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
-        options: ChainOptions.Options,
-        compression: Boolean,
-    ) {
-        if (!this::grpcUpstreamsScheduler.isInitialized) {
-            grpcUpstreamsScheduler = Schedulers.fromExecutorService(
-                Executors.newFixedThreadPool(2),
-                "GrpcUpstreamsStatuses",
-            )
-        }
-        val endpoint = config.connection!!
-        val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}")
-        val ds = GrpcUpstreams(
-            config.id!!,
-            hash,
-            config.role,
-            endpoint.host!!,
-            endpoint.port,
-            endpoint.auth,
-            endpoint.tokenAuth,
-            authorizationConfig,
-            compression,
-            fileResolver,
-            endpoint.upstreamRating,
-            config.labels,
-            grpcUpstreamsScheduler,
-            channelExecutor,
-            chainsConfig,
-            grpcTracing,
-            clientSpansInterceptor,
-            maxMetadataSize,
-            headScheduler,
-            grpcAuthContext,
-        ).apply {
-            timeout = options.timeout
-        }
-        log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
-        ds.start()
-            .doOnNext {
-                log.info("Chain ${it.chain} ${it.type} through gRPC at ${endpoint.host}:${endpoint.port}. With caps: ${it.upstream.getCapabilities()}")
-            }
-            .subscribe(eventPublisher::publishEvent)
-    }
-
-    private fun buildHttpFactory(conn: HttpEndpoint?, urls: ArrayList<URI>? = null): HttpRpcFactory? {
-        return conn?.let { endpoint ->
-            val tls = conn.tls?.let { tls ->
-                tls.ca?.let { ca ->
-                    fileResolver.resolve(ca).readBytes()
-                }
-            }
-            urls?.add(endpoint.url)
-            HttpRpcFactory(endpoint.url.toString(), conn.basicAuth, tls)
-        }
-    }
-
-    private fun buildWsFactory(
-        id: String,
-        chain: Chain,
-        conn: RpcConnection,
-        urls: ArrayList<URI>? = null,
-    ): WsConnectionPoolFactory? {
-        return conn.ws?.let { endpoint ->
-            val wsConnectionFactory = WsConnectionFactory(
-                id,
-                chain,
-                endpoint.url,
-                endpoint.origin ?: URI("http://localhost"),
-                wsScheduler,
-            ).apply {
-                config = endpoint
-                basicAuth = endpoint.basicAuth
-            }
-            val wsApi = WsConnectionPoolFactory(
-                id,
-                endpoint.connections,
-                wsConnectionFactory,
-            )
-            urls?.add(endpoint.url)
-            wsApi
-        }
-    }
-
-    private fun buildConnectorFactory(
-        id: String,
-        conn: RpcConnection,
-        chain: Chain,
-        forkChoice: ForkChoice,
-        blockValidator: BlockValidator,
-        chainsConf: ChainConfig,
-    ): GenericConnectorFactory? {
-        val urls = ArrayList<URI>()
-        val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
-        val httpFactory = buildHttpFactory(conn.rpc, urls)
-        log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
-        val connectorFactory =
-            GenericConnectorFactory(
-                conn.resolveMode(),
-                wsFactoryApi,
-                httpFactory,
-                forkChoice,
-                blockValidator,
-                wsConnectionResubscribeScheduler,
-                headScheduler,
-                headLivenessScheduler,
-                chainsConf.expectedBlockTime,
-            )
-        if (!connectorFactory.isValid()) {
-            log.warn("Upstream configuration is invalid (probably no http endpoint)")
-            return null
-        }
-        return connectorFactory
-    }
-
-    @VisibleForTesting
-    private fun getHash(nodeId: Int?, obj: Any): Byte =
-        nodeId?.toByte() ?: (obj.hashCode() % 255).let {
-            if (it == 0) 1 else it
-        }.let { nonZeroHash ->
-            listOf<Function<Int, Int>>(
-                Function { i -> i },
-                Function { i -> (-i) },
-                Function { i -> 127 - abs(i) },
-                Function { i -> abs(i) - 128 },
-            ).map {
-                it.apply(nonZeroHash).toByte()
-            }.firstOrNull {
-                hashes[it] != true
-            }?.let {
-                hashes[it] = true
-                it
-            } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
-                it != 0 && hashes[it.toByte()] != true
-            }.toByte()
-        }
 }

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

@@ -0,0 +1,77 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.FileResolver
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.foundation.ChainOptions
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.CallTargetsHolder
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.MergedHead
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead
+import io.emeraldpay.dshackle.upstream.bitcoin.EsploraClient
+import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
+import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer
+import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
+import org.springframework.stereotype.Component
+import reactor.core.scheduler.Scheduler
+import java.util.concurrent.atomic.AtomicInteger
+
+@Component
+class BitcoinUpstreamCreator(
+    chainsConfig: ChainsConfig,
+    callTargets: CallTargetsHolder,
+    private val genericConnectorFactoryCreator: ConnectorFactoryCreator,
+    private val fileResolver: FileResolver,
+    private val headScheduler: Scheduler,
+) : UpstreamCreator(chainsConfig, callTargets) {
+    private var seq = AtomicInteger(0)
+
+    override fun createUpstream(
+        upstreamsConfig: UpstreamsConfig.Upstream<*>,
+        chain: Chain,
+        options: ChainOptions.Options,
+        chainConf: ChainsConfig.ChainConfig,
+    ): Upstream? {
+        val config = upstreamsConfig.cast(UpstreamsConfig.BitcoinConnection::class.java)
+        val conn = config.connection!!
+        val httpFactory = genericConnectorFactoryCreator.buildHttpFactory(conn.rpc)
+        if (httpFactory == null) {
+            log.warn("Upstream doesn't have API configuration")
+            return null
+        }
+        val directApi = httpFactory.create(config.id, chain)
+        val esplora = conn.esplora?.let { endpoint ->
+            val tls = endpoint.tls?.let { tls ->
+                tls.ca?.let { ca ->
+                    fileResolver.resolve(ca).readBytes()
+                }
+            }
+            EsploraClient(endpoint.url, endpoint.basicAuth, tls)
+        }
+
+        val extractBlock = ExtractBlock()
+        val rpcHead = BitcoinRpcHead(directApi, extractBlock, headScheduler = headScheduler)
+        val head: Head = conn.zeroMq?.let { zeroMq ->
+            val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock")
+            val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock, headScheduler)
+            MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice(), headScheduler)
+        } ?: rpcHead
+
+        val methods = buildMethods(config, chain)
+        val upstream = BitcoinRpcUpstream(
+            config.id
+                ?: "bitcoin-${seq.getAndIncrement()}",
+            chain, directApi, head,
+            options, config.role,
+            QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)),
+            methods, esplora, chainConf,
+        )
+        upstream.start()
+        return upstream
+    }
+}

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

@@ -0,0 +1,23 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.upstream.BlockValidator
+import io.emeraldpay.dshackle.upstream.HttpRpcFactory
+import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
+import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
+import java.net.URI
+
+interface ConnectorFactoryCreator {
+    fun createConnectorFactoryCreator(
+        id: String,
+        conn: UpstreamsConfig.RpcConnection,
+        chain: Chain,
+        forkChoice: ForkChoice,
+        blockValidator: BlockValidator,
+        chainsConf: ChainsConfig.ChainConfig,
+    ): ConnectorFactory?
+
+    fun buildHttpFactory(conn: UpstreamsConfig.HttpEndpoint?, urls: ArrayList<URI>? = null): HttpRpcFactory?
+}

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

@@ -0,0 +1,36 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.foundation.ChainOptions
+import io.emeraldpay.dshackle.upstream.CallTargetsHolder
+import io.emeraldpay.dshackle.upstream.Upstream
+import org.springframework.stereotype.Component
+
+@Component
+class EthereumUpstreamCreator(
+    chainsConfig: ChainsConfig,
+    callTargets: CallTargetsHolder,
+    genericConnectorFactoryCreator: ConnectorFactoryCreator,
+) : GenericUpstreamCreator(chainsConfig, callTargets, genericConnectorFactoryCreator) {
+
+    override fun createUpstream(
+        upstreamsConfig: UpstreamsConfig.Upstream<*>,
+        chain: Chain,
+        options: ChainOptions.Options,
+        chainConf: ChainsConfig.ChainConfig,
+    ): Upstream? {
+        val posConn = upstreamsConfig.cast(UpstreamsConfig.EthereumPosConnection::class.java)
+
+        return buildGenericUpstream(
+            upstreamsConfig.nodeId,
+            upstreamsConfig,
+            posConn.connection?.execution ?: throw IllegalStateException("Empty execution config"),
+            chain,
+            options,
+            chainConf,
+            posConn.connection?.upstreamRating ?: 0,
+        )
+    }
+}

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

@@ -0,0 +1,98 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.FileResolver
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.upstream.BlockValidator
+import io.emeraldpay.dshackle.upstream.HttpRpcFactory
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionFactory
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
+import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
+import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
+import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Component
+import reactor.core.scheduler.Scheduler
+import java.net.URI
+
+@Component
+class GenericConnectorFactoryCreator(
+    private val fileResolver: FileResolver,
+    private val wsConnectionResubscribeScheduler: Scheduler,
+    private val headScheduler: Scheduler,
+    private val wsScheduler: Scheduler,
+    private val headLivenessScheduler: Scheduler,
+) : ConnectorFactoryCreator {
+    private val log = LoggerFactory.getLogger(this::class.java)
+
+    override fun createConnectorFactoryCreator(
+        id: String,
+        conn: UpstreamsConfig.RpcConnection,
+        chain: Chain,
+        forkChoice: ForkChoice,
+        blockValidator: BlockValidator,
+        chainsConf: ChainsConfig.ChainConfig,
+    ): ConnectorFactory? {
+        val urls = ArrayList<URI>()
+        val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
+        val httpFactory = buildHttpFactory(conn.rpc, urls)
+        log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
+        val connectorFactory =
+            GenericConnectorFactory(
+                conn.resolveMode(),
+                wsFactoryApi,
+                httpFactory,
+                forkChoice,
+                blockValidator,
+                wsConnectionResubscribeScheduler,
+                headScheduler,
+                headLivenessScheduler,
+                chainsConf.expectedBlockTime,
+            )
+        if (!connectorFactory.isValid()) {
+            log.warn("Upstream configuration is invalid (probably no http endpoint)")
+            return null
+        }
+        return connectorFactory
+    }
+
+    override fun buildHttpFactory(conn: UpstreamsConfig.HttpEndpoint?, urls: ArrayList<URI>?): HttpRpcFactory? {
+        return conn?.let { endpoint ->
+            val tls = conn.tls?.let { tls ->
+                tls.ca?.let { ca ->
+                    fileResolver.resolve(ca).readBytes()
+                }
+            }
+            urls?.add(endpoint.url)
+            HttpRpcFactory(endpoint.url.toString(), conn.basicAuth, tls)
+        }
+    }
+
+    private fun buildWsFactory(
+        id: String,
+        chain: Chain,
+        conn: UpstreamsConfig.RpcConnection,
+        urls: ArrayList<URI>? = null,
+    ): WsConnectionPoolFactory? {
+        return conn.ws?.let { endpoint ->
+            val wsConnectionFactory = WsConnectionFactory(
+                id,
+                chain,
+                endpoint.url,
+                endpoint.origin ?: URI("http://localhost"),
+                wsScheduler,
+            ).apply {
+                config = endpoint
+                basicAuth = endpoint.basicAuth
+            }
+            val wsApi = WsConnectionPoolFactory(
+                id,
+                endpoint.connections,
+                wsConnectionFactory,
+            )
+            urls?.add(endpoint.url)
+            wsApi
+        }
+    }
+}

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

@@ -0,0 +1,119 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.foundation.ChainOptions
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.BlockValidator
+import io.emeraldpay.dshackle.upstream.CallTargetsHolder
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
+import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
+import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
+import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory
+import org.springframework.stereotype.Component
+import java.util.function.Function
+import kotlin.math.abs
+
+@Component
+open class GenericUpstreamCreator(
+    chainsConfig: ChainsConfig,
+    callTargets: CallTargetsHolder,
+    private val genericConnectorFactoryCreator: ConnectorFactoryCreator,
+) : UpstreamCreator(chainsConfig, callTargets) {
+    private val hashes: MutableMap<Byte, Boolean> = HashMap()
+
+    override fun createUpstream(
+        upstreamsConfig: UpstreamsConfig.Upstream<*>,
+        chain: Chain,
+        options: ChainOptions.Options,
+        chainConf: ChainsConfig.ChainConfig,
+    ): Upstream? {
+        return buildGenericUpstream(
+            upstreamsConfig.nodeId,
+            upstreamsConfig,
+            upstreamsConfig.connection as UpstreamsConfig.RpcConnection,
+            chain,
+            options,
+            chainConf,
+            0,
+        )
+    }
+
+    protected fun buildGenericUpstream(
+        nodeId: Int?,
+        config: UpstreamsConfig.Upstream<*>,
+        connection: UpstreamsConfig.RpcConnection,
+        chain: Chain,
+        options: ChainOptions.Options,
+        chainConfig: ChainsConfig.ChainConfig,
+        nodeRating: Int,
+    ): Upstream? {
+        if (config.connection == null) {
+            log.warn("Upstream doesn't have connection configuration")
+            return null
+        }
+
+        val cs = ChainSpecificRegistry.resolve(chain)
+
+        val connectorFactory = genericConnectorFactoryCreator.createConnectorFactoryCreator(
+            config.id!!,
+            connection,
+            chain,
+            NoChoiceWithPriorityForkChoice(nodeRating, config.id!!),
+            BlockValidator.ALWAYS_VALID,
+            chainConfig,
+        ) ?: return null
+
+        val methods = buildMethods(config, chain)
+
+        val hashUrl = connection.let {
+            if (it.connectorMode == GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
+        }
+        val hash = getHash(nodeId, hashUrl!!)
+
+        val upstream = GenericUpstream(
+            config.id!!,
+            chain,
+            hash,
+            options,
+            config.role,
+            methods,
+            QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(config.labels)),
+            chainConfig,
+            connectorFactory,
+            cs::validator,
+            cs::labelDetector,
+            cs::subscriptionTopics,
+        )
+
+        upstream.start()
+        if (!upstream.isRunning) {
+            log.debug("Upstream ${upstream.getId()} is not running, it can't be added")
+            return null
+        }
+        return upstream
+    }
+
+    private fun getHash(nodeId: Int?, obj: Any): Byte =
+        nodeId?.toByte() ?: (obj.hashCode() % 255).let {
+            if (it == 0) 1 else it
+        }.let { nonZeroHash ->
+            listOf<Function<Int, Int>>(
+                Function { i -> i },
+                Function { i -> (-i) },
+                Function { i -> 127 - abs(i) },
+                Function { i -> abs(i) - 128 },
+            ).map {
+                it.apply(nonZeroHash).toByte()
+            }.firstOrNull {
+                hashes[it] != true
+            }?.let {
+                hashes[it] = true
+                it
+            } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
+                it != 0 && hashes[it.toByte()] != true
+            }.toByte()
+        }
+}

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

@@ -0,0 +1,67 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.foundation.ChainOptions
+import io.emeraldpay.dshackle.upstream.CallTargetsHolder
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+abstract class UpstreamCreator(
+    private val chainsConfig: ChainsConfig,
+    private val callTargets: CallTargetsHolder,
+) {
+    protected val log: Logger = LoggerFactory.getLogger(this::class.java)
+
+    fun createUpstream(
+        upstreamsConfig: UpstreamsConfig.Upstream<*>,
+        defaultOptions: Map<Chain, ChainOptions.PartialOptions>,
+    ): Upstream? {
+        val chain = Global.chainById(upstreamsConfig.chain)
+        if (chain == Chain.UNSPECIFIED) {
+            throw IllegalArgumentException("Chain is unknown: ${upstreamsConfig.chain}")
+        }
+        val chainConfig = chainsConfig.resolve(upstreamsConfig.chain!!)
+        val options = chainConfig.options
+            .merge(defaultOptions[chain] ?: ChainOptions.PartialOptions.getDefaults())
+            .merge(upstreamsConfig.options ?: ChainOptions.PartialOptions())
+            .buildOptions()
+
+        return createUpstream(upstreamsConfig, chain, options, chainConfig)
+    }
+
+    protected abstract fun createUpstream(
+        upstreamsConfig: UpstreamsConfig.Upstream<*>,
+        chain: Chain,
+        options: ChainOptions.Options,
+        chainConf: ChainsConfig.ChainConfig,
+    ): Upstream?
+
+    protected fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
+        return if (config.methods != null || config.methodGroups != null) {
+            ManagedCallMethods(
+                delegate = callTargets.getDefaultMethods(chain),
+                enabled = config.methods?.enabled?.map { it.name }?.toSet() ?: emptySet(),
+                disabled = config.methods?.disabled?.map { it.name }?.toSet() ?: emptySet(),
+                groupsEnabled = config.methodGroups?.enabled ?: emptySet(),
+                groupsDisabled = config.methodGroups?.disabled ?: emptySet(),
+            ).also {
+                config.methods?.enabled?.forEach { m ->
+                    if (m.quorum != null) {
+                        it.setQuorum(m.name, m.quorum)
+                    }
+                    if (m.static != null) {
+                        it.setStaticResponse(m.name, m.static)
+                    }
+                }
+            }
+        } else {
+            callTargets.getDefaultMethods(chain)
+        }
+    }
+}

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

@@ -0,0 +1,28 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.BlockchainType
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.foundation.ChainOptions
+import io.emeraldpay.dshackle.upstream.Upstream
+import org.springframework.stereotype.Component
+
+@Component
+class UpstreamFactory(
+    private val genericUpstreamCreator: GenericUpstreamCreator,
+    private val ethereumUpstreamCreator: EthereumUpstreamCreator,
+    private val bitcoinUpstreamCreator: BitcoinUpstreamCreator,
+) {
+
+    fun createUpstream(
+        type: BlockchainType,
+        upstreamsConfig: UpstreamsConfig.Upstream<*>,
+        defaultOptions: Map<Chain, ChainOptions.PartialOptions>,
+    ): Upstream? {
+        return when (type) {
+            BlockchainType.ETHEREUM -> ethereumUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions)
+            BlockchainType.BITCOIN -> bitcoinUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions)
+            else -> genericUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions)
+        }
+    }
+}

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

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.foundation.ChainOptions
 import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
@@ -56,7 +57,7 @@ abstract class DefaultUpstream(
     private val statusStream = Sinks.many()
         .multicast()
         .directBestEffort<UpstreamAvailability>()
-    protected val stateStream: Sinks.Many<Boolean> = Sinks.many()
+    protected val stateEventStream: Sinks.Many<UpstreamChangeEvent> = Sinks.many()
         .multicast()
         .directBestEffort()
 
@@ -109,8 +110,8 @@ abstract class DefaultUpstream(
         return statusStream.asFlux().distinctUntilChanged()
     }
 
-    override fun observeState(): Flux<Boolean> {
+    override fun observeState(): Flux<UpstreamChangeEvent> {
-        return stateStream.asFlux()
+        return stateEventStream.asFlux()
     }
 
     override fun setLag(lag: Long) {

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

@@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.CallSelector
 import io.micrometer.core.instrument.Gauge
 import io.micrometer.core.instrument.Meter
 import io.micrometer.core.instrument.Metrics
@@ -34,17 +35,13 @@ import io.micrometer.core.instrument.Tag
 import org.apache.commons.collections4.Factory
 import org.apache.commons.collections4.FunctorException
 import org.slf4j.LoggerFactory
-import org.springframework.context.event.EventListener
-import org.springframework.core.Ordered
-import org.springframework.core.annotation.Order
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.Sinks
+import reactor.core.scheduler.Scheduler
 import java.time.Duration
 import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.locks.ReentrantLock
-import kotlin.concurrent.withLock
 
 /**
  * Aggregation of multiple upstreams responding to a single blockchain
@@ -52,8 +49,9 @@ import kotlin.concurrent.withLock
 abstract class Multistream(
     val chain: Chain,
     val caches: Caches,
+    val callSelector: CallSelector?,
+    multistreamEventsScheduler: Scheduler,
 ) : Upstream, Lifecycle {
-
     abstract fun getUpstreams(): MutableList<out Upstream>
     abstract fun addUpstreamInternal(u: Upstream)
 
@@ -66,8 +64,6 @@ abstract class Multistream(
     private var started = false
 
     private var cacheSubscription: Disposable? = null
-    private val reconfigLock = ReentrantLock()
-    private val eventLock = ReentrantLock()
 
     @Volatile
     private var callMethods: CallMethods? = null
@@ -94,6 +90,9 @@ abstract class Multistream(
     private val updateUpstreams = Sinks.many()
         .multicast()
         .directBestEffort<Upstream>()
+    private val upstreamsSink = Sinks.many()
+        .multicast()
+        .directBestEffort<UpstreamChangeEvent>()
 
     override fun getSubscriptionTopics(): List<String> {
         return getEgressSubscription().getAvailableTopics()
@@ -127,10 +126,6 @@ abstract class Multistream(
         )
     }
 
-    open fun init() {
-        onUpstreamsUpdated()
-    }
-
     init {
         UpstreamAvailability.entries.forEach { status ->
             Metrics.gauge(
@@ -149,6 +144,12 @@ abstract class Multistream(
         ) {
             getAll().size.toDouble()
         }
+
+        upstreamsSink.asFlux()
+            .publishOn(multistreamEventsScheduler)
+            .subscribe {
+                onUpstreamChange(it)
+            }
     }
 
     /**
@@ -206,36 +207,34 @@ abstract class Multistream(
         throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")
     }
 
-    open fun onUpstreamsUpdated() {
+    protected open fun onUpstreamsUpdated() {
-        reconfigLock.withLock {
+        val upstreams = getAll()
-            val upstreams = getAll()
+        val availableUpstreams = upstreams.filter { it.isAvailable() }
-            val availableUpstreams = upstreams.filter { it.isAvailable() }
+        availableUpstreams.map { it.getMethods() }.let {
-            availableUpstreams.map { it.getMethods() }.let {
+            callMethods = AggregatedCallMethods(it)
-                callMethods = AggregatedCallMethods(it)
+        }
-            }
+        capabilities = if (upstreams.isEmpty()) {
-            capabilities = if (upstreams.isEmpty()) {
+            emptySet()
-                emptySet()
+        } else {
-            } else {
+            availableUpstreams.map { up ->
-                availableUpstreams.map { up ->
+                up.getCapabilities()
-                    up.getCapabilities()
+            }.let {
-                }.let {
+                if (it.isNotEmpty()) {
-                    if (it.isNotEmpty()) {
+                    it.reduce { acc, curr -> acc + curr }
-                        it.reduce { acc, curr -> acc + curr }
+                } else {
-                    } else {
+                    emptySet()
-                        emptySet()
-                    }
                 }
             }
-            quorumLabels = getQuorumLabels(availableUpstreams)
+        }
-            when {
+        quorumLabels = getQuorumLabels(availableUpstreams)
-                upstreams.size == 1 -> {
+        when {
-                    lagObserver?.stop()
+            upstreams.size == 1 -> {
-                    lagObserver = null
+                lagObserver?.stop()
-                    upstreams[0].setLag(0)
+                lagObserver = null
-                }
+                upstreams[0].setLag(0)
+            }
 
-                upstreams.size > 1 -> if (lagObserver == null) lagObserver = makeLagObserver()
+            upstreams.size > 1 -> if (lagObserver == null) lagObserver = makeLagObserver()
-            }
         }
     }
 
@@ -265,7 +264,7 @@ abstract class Multistream(
         ).distinct()
     }
 
-    override fun observeState(): Flux<Boolean> {
+    override fun observeState(): Flux<UpstreamChangeEvent> {
         return Flux.empty()
     }
 
@@ -319,8 +318,9 @@ abstract class Multistream(
             .distinctUntilChanged {
                 it.getId()
             }.flatMap { upstream ->
-                val statusStream = upstream.observeStatus().map { upstream }
+                val statusStream = upstream.observeStatus()
-                val stateStream = upstream.observeState().map { upstream }
+                    .map { UpstreamChangeEvent(this.chain, upstream, UpstreamChangeEvent.ChangeType.UPDATED) }
+                val stateStream = upstream.observeState()
                 Flux.merge(stateStream, statusStream)
                     .takeUntilOther(
                         subscribeRemovedUpstreams()
@@ -330,7 +330,7 @@ abstract class Multistream(
                     )
             }
             .subscribe {
-                onUpstreamChange(UpstreamChangeEvent(this.chain, it, UpstreamChangeEvent.ChangeType.UPDATED))
+                this.processUpstreamsEvents(it)
             }
     }
 
@@ -350,11 +350,9 @@ abstract class Multistream(
     }
 
     fun onHeadUpdated(head: Head) {
-        reconfigLock.withLock {
+        cacheSubscription?.dispose()
-            cacheSubscription?.dispose()
+        cacheSubscription = head.getFlux().subscribe {
-            cacheSubscription = head.getFlux().subscribe {
+            caches.cache(Caches.Tag.LATEST, it)
-                caches.cache(Caches.Tag.LATEST, it)
-            }
         }
         caches.setHead(head)
     }
@@ -418,47 +416,49 @@ abstract class Multistream(
         return event.chain == this.chain
     }
 
-    @EventListener
+    fun processUpstreamsEvents(event: UpstreamChangeEvent) {
-    @Order(Ordered.HIGHEST_PRECEDENCE)
+        upstreamsSink.emitNext(
-    fun onUpstreamChange(event: UpstreamChangeEvent) {
+            event,
+        ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+    }
+
+    private fun onUpstreamChange(event: UpstreamChangeEvent) {
         val chain = event.chain
         if (this.chain == chain) {
-            eventLock.withLock {
+            log.debug("Processing event {}", event)
-                log.debug("Processing event $event")
+            when (event.type) {
-                when (event.type) {
+                UpstreamChangeEvent.ChangeType.REVALIDATED -> {}
-                    UpstreamChangeEvent.ChangeType.REVALIDATED -> {}
+                UpstreamChangeEvent.ChangeType.UPDATED -> {
-                    UpstreamChangeEvent.ChangeType.UPDATED -> {
+                    onUpstreamsUpdated()
-                        onUpstreamsUpdated()
+                    updateUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
-                        updateUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                }
-                    }
 
-                    UpstreamChangeEvent.ChangeType.ADDED -> {
+                UpstreamChangeEvent.ChangeType.ADDED -> {
-                        if (!started) {
+                    if (!started) {
-                            start()
+                        start()
-                        }
+                    }
-                        if (event.upstream is CachesEnabled) {
+                    if (event.upstream is CachesEnabled) {
-                            event.upstream.setCaches(caches)
+                        event.upstream.setCaches(caches)
-                        }
+                    }
-                        addUpstream(event.upstream).takeIf { it }?.let {
+                    addUpstream(event.upstream).takeIf { it }?.let {
-                            try {
+                        try {
-                                addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                            addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
-                                onUpstreamsUpdated()
+                            onUpstreamsUpdated()
-                                log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
+                            log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
-                            } catch (e: Sinks.EmissionException) {
+                        } catch (e: Sinks.EmissionException) {
-                                log.error("error during event processing $event", e)
+                            log.error("error during event processing $event", e)
-                            }
                         }
                     }
+                }
 
-                    UpstreamChangeEvent.ChangeType.REMOVED -> {
+                UpstreamChangeEvent.ChangeType.REMOVED -> {
-                        removeUpstream(event.upstream.getId()).takeIf { it }?.let {
+                    removeUpstream(event.upstream.getId()).takeIf { it }?.let {
-                            try {
+                        try {
-                                removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                            removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
-                                onUpstreamsUpdated()
+                            onUpstreamsUpdated()
-                                log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
+                            log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
-                            } catch (e: Sinks.EmissionException) {
+                        } catch (e: Sinks.EmissionException) {
-                                log.error("error during event processing $event", e)
+                            log.error("error during event processing $event", e)
-                            }
                         }
                     }
                 }

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

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.upstream
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.foundation.ChainOptions
 import io.emeraldpay.dshackle.reader.JsonRpcReader
+import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import reactor.core.publisher.Flux
 
@@ -26,7 +27,7 @@ interface Upstream : Lifecycle {
     fun isAvailable(): Boolean
     fun getStatus(): UpstreamAvailability
     fun observeStatus(): Flux<UpstreamAvailability>
-    fun observeState(): Flux<Boolean>
+    fun observeState(): Flux<UpstreamChangeEvent>
     fun getHead(): Head
 
     /**

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt

@@ -40,10 +40,11 @@ import reactor.core.scheduler.Scheduler
 @Suppress("UNCHECKED_CAST")
 open class BitcoinMultistream(
     chain: Chain,
+    multistreamEventsScheduler: Scheduler,
     private val sourceUpstreams: MutableList<BitcoinUpstream>,
     caches: Caches,
     private val headScheduler: Scheduler,
-) : Multistream(chain, caches), Lifecycle {
+) : Multistream(chain, caches, null, multistreamEventsScheduler), Lifecycle {
 
     private var head: Head = EmptyHead()
     private var esplora = sourceUpstreams.find { it.esploraClient != null }?.esploraClient
@@ -59,13 +60,6 @@ open class BitcoinMultistream(
         sourceUpstreams.add(u as BitcoinUpstream)
     }
 
-    override fun init() {
-        if (sourceUpstreams.size > 0) {
-            head = updateHead()
-        }
-        super.init()
-    }
-
     open fun getXpubAddresses(): XpubAddresses? {
         return xpubAddresses
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallSelector.kt

@@ -0,0 +1,10 @@
+package io.emeraldpay.dshackle.upstream.calls
+
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
+import reactor.core.publisher.Mono
+
+interface CallSelector {
+
+    fun getMatcher(method: String, params: String, head: Head, passthrough: Boolean): Mono<Selector.Matcher>
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt

@@ -32,7 +32,7 @@ import java.util.Objects
  */
 class EthereumCallSelector(
     private val caches: Caches,
-) {
+) : CallSelector {
 
     companion object {
         private val log = LoggerFactory.getLogger(EthereumCallSelector::class.java)
@@ -68,7 +68,7 @@ class EthereumCallSelector(
      * @param method JSON RPC name
      * @param params JSON-encoded list of parameters for the method
      */
-    fun getMatcher(method: String, params: String, head: Head, passthrough: Boolean): Mono<Selector.Matcher> {
+    override fun getMatcher(method: String, params: String, head: Head, passthrough: Boolean): Mono<Selector.Matcher> {
         if (method in DefaultEthereumMethods.withFilterIdMethods) {
             return sameUpstreamMatcher(params)
         } else if (!passthrough) { // passthrough indicates we should match only labels

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt

@@ -1,6 +1,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.foundation.ChainOptions.Options
@@ -15,6 +16,8 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamValidator
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.CallSelector
+import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription
@@ -99,4 +102,8 @@ object EthereumChainSpecific : ChainSpecific {
     override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription {
         return EthereumWsIngressSubscription(ws)
     }
+
+    override fun callSelector(caches: Caches): CallSelector {
+        return EthereumCallSelector(caches)
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamValidator
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.CallSelector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
 import io.emeraldpay.dshackle.upstream.polkadot.PolkadotChainSpecific
@@ -58,6 +59,8 @@ interface ChainSpecific {
     fun subscriptionTopics(upstream: GenericUpstream): List<String>
 
     fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription
+
+    fun callSelector(caches: Caches): CallSelector?
 }
 
 object ChainSpecificRegistry {

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt

@@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Selector.Matcher
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.calls.CallSelector
 import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
 import org.springframework.util.ConcurrentReferenceHashMap
@@ -44,13 +45,15 @@ import reactor.core.scheduler.Scheduler
 
 open class GenericMultistream(
     chain: Chain,
+    multistreamEventsScheduler: Scheduler,
+    callSelector: CallSelector?,
     private val upstreams: MutableList<Upstream>,
     caches: Caches,
     private val headScheduler: Scheduler,
     cachingReaderBuilder: CachingReaderBuilder,
     private val localReaderBuilder: LocalReaderBuilder,
     private val subscriptionBuilder: SubscriptionBuilder,
-) : Multistream(chain, caches) {
+) : Multistream(chain, caches, callSelector, multistreamEventsScheduler) {
 
     private val cachingReader = cachingReaderBuilder(this, caches, getMethodsFactory())
 
@@ -68,19 +71,8 @@ open class GenericMultistream(
         headScheduler,
     )
 
-    init {
-        this.init()
-    }
-
     private var subscription: EgressSubscription = subscriptionBuilder(this)
 
-    override fun init() {
-        if (upstreams.size > 0) {
-            upstreams.forEach { addHead(it) }
-        }
-        super.init()
-    }
-
     private val filteredHeads: MutableMap<String, Head> =
         ConcurrentReferenceHashMap(16, WEAK)
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt

@@ -21,7 +21,6 @@ import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
 import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector
-import org.springframework.context.ApplicationEventPublisher
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
@@ -39,7 +38,6 @@ open class GenericUpstream(
     private val node: QuorumForLabels.QuorumItem?,
     chainConfig: ChainsConfig.ChainConfig,
     connectorFactory: ConnectorFactory,
-    private val eventPublisher: ApplicationEventPublisher?,
     validatorBuilder: UpstreamValidatorBuilder,
     labelsDetectorBuilder: LabelsDetectorBuilder,
     private val subscriptionTopics: (GenericUpstream) -> List<String>,
@@ -132,13 +130,9 @@ open class GenericUpstream(
 
                     ValidateUpstreamSettingsResult.UPSTREAM_VALID -> {
                         upstreamStart()
-                        eventPublisher?.publishEvent(
+                        stateEventStream.emitNext(
-                            UpstreamChangeEvent(
+                            UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.ADDED),
-                                chain,
+                        ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
-                                this,
-                                UpstreamChangeEvent.ChangeType.ADDED,
-                            ),
-                        )
                         disposeValidationSettingsSubscription()
                     }
 
@@ -166,7 +160,9 @@ open class GenericUpstream(
         }
         livenessSubscription = connector.hasLiveSubscriptionHead().subscribe({
             hasLiveSubscriptionHead.set(it)
-            stateStream.emitNext(true) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+            stateEventStream.emitNext(
+                UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED),
+            ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
         }, {
             log.debug("Error while checking live subscription for ${getId()}", it)
         },)

src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.annotation.JsonProperty
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
@@ -19,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.NoopCachingReader
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamValidator
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.CallSelector
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
 import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder
 import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
@@ -104,6 +106,10 @@ object PolkadotChainSpecific : ChainSpecific {
     override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription {
         return PolkadotIngressSubscription(ws)
     }
+
+    override fun callSelector(caches: Caches): CallSelector? {
+        return null
+    }
 }
 
 @JsonIgnoreProperties(ignoreUnknown = true)

src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.annotation.JsonProperty
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
@@ -21,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.NoopCachingReader
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamValidator
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.calls.CallSelector
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
 import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder
 import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
@@ -102,6 +104,10 @@ object StarknetChainSpecific : ChainSpecific {
     override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription {
         return NoIngressSubscription()
     }
+
+    override fun callSelector(caches: Caches): CallSelector? {
+        return null
+    }
 }
 
 @JsonIgnoreProperties(ignoreUnknown = true)

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy

@@ -28,11 +28,13 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.reader.RpcReader
 import io.emeraldpay.dshackle.reader.RpcReaderFactory
+import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.test.MultistreamHolderMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
+import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -601,9 +603,10 @@ class NativeCallSpec extends Specification {
                 new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET),
                 [] as Set, [] as Set, ["filter"] as Set, [] as Set
         )
-        def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, TestingCommons.upstream(
+        def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, new ArrayList<GenericUpstream>())
-                TestingCommons.api(), methods
+        multistream.processUpstreamsEvents(
-        ))
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, TestingCommons.upstream(TestingCommons.api(), methods), UpstreamChangeEvent.ChangeType.ADDED)
+        )
         multistream.customHead = Mock(Head)
         def multistreamHolder = Mock(MultistreamHolder) {
             _ * it.observeChains() >> Flux.empty()
@@ -636,9 +639,10 @@ class NativeCallSpec extends Specification {
                 new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET),
                 [] as Set, [] as Set, ["filter"] as Set, [] as Set
         )
-        def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, TestingCommons.upstream(
+        def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM__MAINNET, new ArrayList<GenericUpstream>())
-                TestingCommons.api(), methods
+        multistream.processUpstreamsEvents(
-        ))
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, TestingCommons.upstream(TestingCommons.api(), methods), UpstreamChangeEvent.ChangeType.ADDED)
+        )
         multistream.customHead = Mock(Head)
         def multistreamHolder = Mock(MultistreamHolder) {
             _ * it.observeChains() >> Flux.empty()

src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy

@@ -1,191 +0,0 @@
-package io.emeraldpay.dshackle.startup
-
-import brave.Tracing
-import brave.grpc.GrpcTracing
-import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.FileResolver
-import io.emeraldpay.dshackle.config.AuthorizationConfig
-import io.emeraldpay.dshackle.config.ChainsConfig
-import io.emeraldpay.dshackle.config.CompressionConfig
-import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.quorum.NotNullQuorum
-import io.emeraldpay.dshackle.upstream.CallTargetsHolder
-import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
-import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext
-import org.springframework.context.ApplicationEventPublisher
-import reactor.core.scheduler.Schedulers
-import spock.lang.Specification
-
-import java.util.concurrent.Executors
-
-class ConfiguredUpstreamsSpec extends Specification {
-
-    def "Applied  quorum to extra methods"() {
-        setup:
-        def callTargetsHolder = new CallTargetsHolder()
-        def configurer = new ConfiguredUpstreams(
-                Stub(FileResolver),
-                new UpstreamsConfig(),
-                Stub(CompressionConfig),
-                callTargetsHolder,
-                Mock(ApplicationEventPublisher),
-                Executors.newFixedThreadPool(1),
-                ChainsConfig.default(),
-                GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.boundedElastic(),
-                null,
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                AuthorizationConfig.default(),
-                new GrpcAuthContext()
-        )
-        def methods = new UpstreamsConfig.Methods(
-                [
-                        new UpstreamsConfig.Method("foo_bar", null, null),
-                        new UpstreamsConfig.Method("foo_bar", "not_empty", null)
-                ] as Set,
-                [] as Set
-        )
-        def upstream = new UpstreamsConfig.Upstream()
-        upstream.methods = methods
-        when:
-        def act = configurer.buildMethods(upstream, Chain.ETHEREUM__MAINNET)
-        then:
-        act instanceof ManagedCallMethods
-        act.createQuorumFor("foo_bar") instanceof NotNullQuorum
-    }
-
-    def "Got static response from extra methods"() {
-        setup:
-        def callTargetsHolder = new CallTargetsHolder()
-        def configurer = new ConfiguredUpstreams(
-                Stub(FileResolver),
-                new UpstreamsConfig(),
-                Stub(CompressionConfig),
-                callTargetsHolder,
-                Mock(ApplicationEventPublisher),
-                Executors.newFixedThreadPool(1),
-                ChainsConfig.default(),
-                GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.boundedElastic(),
-                null,
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                AuthorizationConfig.default(),
-                new GrpcAuthContext()
-        )
-        def methods = new UpstreamsConfig.Methods(
-                [
-                        new UpstreamsConfig.Method("foo_bar", null, "static_response")
-                ] as Set,
-                [] as Set
-        )
-        def upstream = new UpstreamsConfig.Upstream()
-        upstream.methods = methods
-        when:
-        def act = configurer.buildMethods(upstream, Chain.ETHEREUM__MAINNET)
-        then:
-        act instanceof ManagedCallMethods
-        new String(act.executeHardcoded("foo_bar")) == "\"static_response\""
-    }
-
-    def "Calculate node-id"() {
-        setup:
-        def callTargetsHolder = new CallTargetsHolder()
-        def configurer = new ConfiguredUpstreams(
-                Stub(FileResolver),
-                new UpstreamsConfig(),
-                Stub(CompressionConfig),
-                callTargetsHolder,
-                Mock(ApplicationEventPublisher),
-                Executors.newFixedThreadPool(1),
-                ChainsConfig.default(),
-                GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.boundedElastic(),
-                null,
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                AuthorizationConfig.default(),
-                new GrpcAuthContext()
-        )
-        expect:
-        configurer.getHash(node, src) == expected
-
-        where:
-        node | src | expected
-        1 | "" | 1
-        9 | "hohoho" | 9
-        null | "hohoho" | 120
-    }
-
-    def "Calculate node-id conflicting results"() {
-        setup:
-        def callTargetsHolder = new CallTargetsHolder()
-        def configurer = new ConfiguredUpstreams(
-                Stub(FileResolver),
-                new UpstreamsConfig(),
-                Stub(CompressionConfig),
-                callTargetsHolder,
-                Mock(ApplicationEventPublisher),
-                Executors.newFixedThreadPool(1),
-                ChainsConfig.default(),
-                GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.boundedElastic(),
-                null,
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                AuthorizationConfig.default(),
-                new GrpcAuthContext()
-        )
-        when:
-        def h1 = configurer.getHash(null, "hohoho")
-        def h2 = configurer.getHash(null, "hohoho")
-        def h3 = configurer.getHash(null, "hohoho")
-        def h4 = configurer.getHash(null, "hohoho")
-        def h5 = configurer.getHash(null, "hohoho")
-
-        then:
-        h1 == (byte)120
-        h2 == (byte)-120
-        h3 == (byte)-9
-        h4 == (byte)8
-        h5 == (byte)-128
-    }
-
-    def "Supporting method groups"() {
-        setup:
-        def callTargetsHolder = new CallTargetsHolder()
-        def configurer = new ConfiguredUpstreams(
-                Stub(FileResolver),
-                new UpstreamsConfig(),
-                Stub(CompressionConfig),
-                callTargetsHolder,
-                Mock(ApplicationEventPublisher),
-                Executors.newFixedThreadPool(1),
-                ChainsConfig.default(),
-                GrpcTracing.create(Tracing.newBuilder().build()),
-                Schedulers.boundedElastic(),
-                null,
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                Schedulers.boundedElastic(),
-                AuthorizationConfig.default(),
-                new GrpcAuthContext()
-        )
-        def methodsGroup = new UpstreamsConfig.MethodGroups(
-                ["filter"] as Set,
-                [] as Set
-        )
-        def upstream = new UpstreamsConfig.Upstream()
-        upstream.methodGroups = methodsGroup
-        when:
-        def act = configurer.buildMethods(upstream, Chain.ETHEREUM__MAINNET)
-        then:
-        act instanceof ManagedCallMethods
-        act.supportedMethods.findAll {it.containsIgnoreCase("filter")}.size() == 6
-    }
-}

src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy

@@ -72,7 +72,6 @@ class GenericUpstreamMock extends GenericUpstream {
                 new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)),
                 ChainConfig.default(),
                 new ConnectorFactoryMock(api, new EthereumHeadMock()),
-                null,
                 io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&validator,
                 io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&labelDetector,
                 io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&subscriptionTopics,

src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy

@@ -47,12 +47,13 @@ class MultistreamHolderMock implements MultistreamHolder {
                     upstreams[chain] = up
                 } else if (up instanceof GenericUpstream) {
                     upstreams[chain] = new GenericMultistream(
-                            chain, [up as GenericUpstream], Caches.default(),
+                            chain, Schedulers.immediate(), null, new ArrayList<Upstream>(), Caches.default(),
                             Schedulers.boundedElastic(),
                             EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                             EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                             io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic())
                     )
+                    upstreams[chain].addUpstream(up)
                 } else {
                     throw new IllegalArgumentException("Unsupported upstream type ${up.class}")
                 }
@@ -100,7 +101,7 @@ class MultistreamHolderMock implements MultistreamHolder {
         Head customHead = null
 
         EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<GenericUpstream> upstreams, @NotNull Caches caches) {
-            super(chain, upstreams, caches, Schedulers.boundedElastic(),
+            super(chain, Schedulers.immediate(), null, upstreams, caches, Schedulers.boundedElastic(),
                     EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(new BraveTracer(null, null, null)),
                     EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                     EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))

src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy

@@ -25,15 +25,17 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.EmptyReader
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific
+import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
 import io.emeraldpay.dshackle.upstream.generic.GenericMultistream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific
 import io.emeraldpay.etherjar.domain.BlockHash
-import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
 import io.emeraldpay.etherjar.domain.TransactionId
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import io.micrometer.core.instrument.MeterRegistry
@@ -91,13 +93,15 @@ class TestingCommons {
     }
 
     static Multistream multistream(GenericUpstreamMock up) {
-        return new GenericMultistream(Chain.ETHEREUM__MAINNET, [up], Caches.default(),
+        return new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<Upstream>(), Caches.default(),
                 Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                 EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()),
         ).tap {
-            start()
+            it.processUpstreamsEvents(
+                    new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED)
+            )
         }
     }
 
@@ -117,14 +121,14 @@ class TestingCommons {
     }
 
     static Multistream multistreamWithoutUpstreams(Chain chain) {
-        return new GenericMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(),
+        return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                 EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
     }
 
     static Multistream multistreamClassicWithoutUpstreams(Chain chain) {
-        return new GenericMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(),
+        return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                 EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))

src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy

@@ -28,7 +28,7 @@ class CurrentMultistreamHolderSpec extends Specification {
         def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams())
         def up = new GenericUpstreamMock("test", Chain.ETHEREUM__MAINNET, TestingCommons.api())
         when:
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up, UpstreamChangeEvent.ChangeType.ADDED))
         then:
         current.getAvailable() == [Chain.ETHEREUM__MAINNET]
         current.getUpstream(Chain.ETHEREUM__MAINNET).getAll()[0] == up
@@ -41,10 +41,10 @@ class CurrentMultistreamHolderSpec extends Specification {
         def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM_CLASSIC__MAINNET, TestingCommons.api())
         def up3 = new GenericUpstreamMock("test3", Chain.ETHEREUM__MAINNET, TestingCommons.api())
         when:
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED))
-        current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED))
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED))
-        current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED))
         then:
         current.getAvailable().toSet() == [Chain.ETHEREUM__MAINNET, Chain.ETHEREUM_CLASSIC__MAINNET].toSet()
         current.getUpstream(Chain.ETHEREUM__MAINNET).getAll().toSet() == [up1, up3].toSet()
@@ -59,10 +59,10 @@ class CurrentMultistreamHolderSpec extends Specification {
         def up3 = new GenericUpstreamMock("test3", Chain.ETHEREUM__MAINNET, TestingCommons.api())
         def up1_del = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api())
         when:
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED))
-        current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM_CLASSIC__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM_CLASSIC__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED))
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED))
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1_del, UpstreamChangeEvent.ChangeType.REMOVED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1_del, UpstreamChangeEvent.ChangeType.REMOVED))
         then:
         current.getAvailable().toSet() == [Chain.ETHEREUM__MAINNET, Chain.ETHEREUM_CLASSIC__MAINNET].toSet()
         current.getUpstream(Chain.ETHEREUM__MAINNET).getAll().toSet() == [up3].toSet()
@@ -80,7 +80,7 @@ class CurrentMultistreamHolderSpec extends Specification {
         !act
 
         when:
-        current.getUpstream(Chain.ETHEREUM__MAINNET).onUpstreamChange(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED))
+        current.getUpstream(Chain.ETHEREUM__MAINNET).processUpstreamsEvents(new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED))
         act = current.isAvailable(Chain.ETHEREUM__MAINNET)
 
         then:

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy

@@ -77,7 +77,6 @@ class FilteredApisSpec extends Specification {
                     new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)),
                     ChainsConfig.ChainConfig.default(),
                     connectorFactory,
-                    null,
                     cs.&validator,
                     cs.&labelDetector,
                     cs.&subscriptionTopics,

src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy

@@ -55,7 +55,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
         def up2 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"]))
-        def aggr = new GenericMultistream(Chain.ETHEREUM__MAINNET, [up1, up2], Caches.default(),
+        def aggr = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, [up1, up2], Caches.default(),
                 Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
@@ -190,7 +190,7 @@ class MultistreamSpec extends Specification {
         def up1 = TestingCommons.upstream("test-1", "internal")
         def up2 = TestingCommons.upstream("test-2", "external")
         def up3 = TestingCommons.upstream("test-3", "external")
-        def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, [up1, up2, up3], Caches.default(),
+        def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, [up1, up2, up3], Caches.default(),
                 Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
@@ -263,16 +263,16 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"]))
         def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
-        def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, new ArrayList<GenericMultistream>(), Caches.default(),
+        def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<GenericMultistream>(), Caches.default(),
                 Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                 EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
         when:
-        ms.onUpstreamChange(
+        ms.processUpstreamsEvents(
                 new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)
         )
-        ms.onUpstreamChange(
+        ms.processUpstreamsEvents(
                 new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)
         )
         up1.onStatus(status(Common.AvailabilityEnum.AVAIL_UNAVAILABLE))
@@ -294,7 +294,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"]))
         def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
-        def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, new ArrayList<GenericMultistream>(), Caches.default(),
+        def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<GenericMultistream>(), Caches.default(),
                 Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
@@ -303,10 +303,10 @@ class MultistreamSpec extends Specification {
         def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b")
         def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b")
         when:
-        ms.onUpstreamChange(
+        ms.processUpstreamsEvents(
                 new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)
         )
-        ms.onUpstreamChange(
+        ms.processUpstreamsEvents(
                 new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)
         )
         def head = ms.getHead()
@@ -329,25 +329,32 @@ class MultistreamSpec extends Specification {
         def up1 = TestingCommons.upstream("test-1", "internal")
         def up2 = TestingCommons.upstream("test-2", "external")
         def up3 = TestingCommons.upstream("test-3", "external")
-        def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, [up1, up2, up3], Caches.default(),
+        def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<Upstream>(), Caches.default(),
                 Schedulers.boundedElastic(),
                 EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                 EthereumChainSpecific.INSTANCE.&localReaderBuilder,
                 EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
-        def observer = multistream.lagObserver
+        multistream.processUpstreamsEvents(
-        multistream.onUpstreamsUpdated()
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)
+        )
+        multistream.processUpstreamsEvents(
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.ADDED)
+        )
+        multistream.processUpstreamsEvents(
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up3, UpstreamChangeEvent.ChangeType.ADDED)
+        )
 
         expect:
         multistream.getAll().size() == 3
-        observer.isRunning()
+        multistream.lagObserver.isRunning()
 
-        multistream.getAll().with {
+        multistream.processUpstreamsEvents(
-            remove(0)
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED)
-            remove(1)
+        )
-        }
+        multistream.processUpstreamsEvents(
+                new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.REMOVED)
+        )
         multistream.getAll().size() == 1
-        multistream.onUpstreamsUpdated()
-        !observer.isRunning()
         multistream.lagObserver == null
     }
 
@@ -360,7 +367,7 @@ class MultistreamSpec extends Specification {
     class TestEthereumPosMultistream extends GenericMultistream {
 
         TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<GenericUpstream> upstreams, @NotNull Caches caches) {
-            super(chain, upstreams, caches,
+            super(chain, Schedulers.immediate(), null, upstreams, caches,
                     Schedulers.boundedElastic(),
                     EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
                     EthereumChainSpecific.INSTANCE.&localReaderBuilder,
@@ -381,11 +388,6 @@ class MultistreamSpec extends Specification {
         public <T extends Upstream> T cast(Class<T> selfType) {
             return this
         }
-
-        @Override
-        void init() {
-
-        }
     }
 
     BlockContainer createBlock(long number, String hash) {

src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt

@@ -16,15 +16,12 @@ import io.emeraldpay.dshackle.upstream.Upstream
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
-import org.mockito.ArgumentCaptor
 import org.mockito.kotlin.any
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.never
-import org.mockito.kotlin.times
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
-import org.springframework.context.ApplicationEventPublisher
 import org.springframework.util.ResourceUtils
 import sun.misc.Signal
 import java.io.File
@@ -38,7 +35,6 @@ class ReloadConfigTest {
 
     private val config = mock<Config>()
     private val reloadConfigService = ReloadConfigService(config, fileResolver, mainConfig)
-    private val applicationEventPublisher = mock<ApplicationEventPublisher>()
     private val configuredUpstreams = mock<ConfiguredUpstreams>()
 
     @BeforeEach
@@ -66,7 +62,6 @@ class ReloadConfigTest {
             on { getUpstream(POLYGON__MAINNET) } doReturn msPoly
         }
         val reloadConfigUpstreamService = ReloadConfigUpstreamService(
-            applicationEventPublisher,
             currentMultistreamHolder,
             configuredUpstreams,
         )
@@ -79,8 +74,8 @@ class ReloadConfigTest {
 
         reloadConfig.handle(Signal("HUP"))
 
-        val captor = ArgumentCaptor.forClass(UpstreamChangeEvent::class.java)
+        verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED))
-        verify(applicationEventPublisher, times(2)).publishEvent(captor.capture())
+        verify(msPoly).processUpstreamsEvents(UpstreamChangeEvent(POLYGON__MAINNET, up3, UpstreamChangeEvent.ChangeType.REMOVED))
         verify(configuredUpstreams).processUpstreams(
             UpstreamsConfig(
                 newConfig.defaultOptions,
@@ -90,14 +85,6 @@ class ReloadConfigTest {
 
         assertEquals(3, mainConfig.upstreams!!.upstreams.size)
         assertEquals(newConfig, mainConfig.upstreams)
-        assertEquals(
-            UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED),
-            captor.allValues[0],
-        )
-        assertEquals(
-            UpstreamChangeEvent(POLYGON__MAINNET, up3, UpstreamChangeEvent.ChangeType.REMOVED),
-            captor.allValues[1],
-        )
     }
 
     @Test
@@ -123,7 +110,6 @@ class ReloadConfigTest {
             on { getUpstream(POLYGON__MAINNET) } doReturn msPoly
         }
         val reloadConfigUpstreamService = ReloadConfigUpstreamService(
-            applicationEventPublisher,
             currentMultistreamHolder,
             configuredUpstreams,
         )
@@ -135,8 +121,8 @@ class ReloadConfigTest {
 
         reloadConfig.handle(Signal("HUP"))
 
-        val captor = ArgumentCaptor.forClass(UpstreamChangeEvent::class.java)
+        verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED))
-        verify(applicationEventPublisher, times(2)).publishEvent(captor.capture())
+        verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.REMOVED))
         verify(configuredUpstreams).processUpstreams(
             UpstreamsConfig(
                 newConfig.defaultOptions,
@@ -146,14 +132,6 @@ class ReloadConfigTest {
         verify(msEth).stop()
         assertEquals(1, mainConfig.upstreams!!.upstreams.size)
         assertEquals(newConfig, mainConfig.upstreams)
-        assertEquals(
-            UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED),
-            captor.allValues[0],
-        )
-        assertEquals(
-            UpstreamChangeEvent(ETHEREUM__MAINNET, up2, UpstreamChangeEvent.ChangeType.REMOVED),
-            captor.allValues[1],
-        )
     }
 
     @Test

src/test/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactoryTest.kt

@@ -0,0 +1,62 @@
+package io.emeraldpay.dshackle.startup.configure
+
+import io.emeraldpay.dshackle.BlockchainType
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.Arguments
+import org.junit.jupiter.params.provider.MethodSource
+import org.mockito.Mockito.reset
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.verify
+
+class UpstreamFactoryTest {
+
+    @BeforeEach
+    fun beforeEach() {
+        reset(genericUpstreamCreator, ethereumUpstreamCreator, bitcoinUpstreamCreator)
+    }
+
+    @ParameterizedTest
+    @MethodSource("data")
+    fun `create upstream based on chain`(
+        blockchain: BlockchainType,
+        verify: Runnable,
+    ) {
+        upstreamFactory.createUpstream(blockchain, UpstreamsConfig.Upstream<UpstreamsConfig.UpstreamConnection>(), emptyMap())
+
+        verify.run()
+    }
+
+    companion object {
+        private val genericUpstreamCreator = mock<GenericUpstreamCreator>()
+        private val ethereumUpstreamCreator = mock<EthereumUpstreamCreator>()
+        private val bitcoinUpstreamCreator = mock<BitcoinUpstreamCreator>()
+        private val upstreamFactory = UpstreamFactory(genericUpstreamCreator, ethereumUpstreamCreator, bitcoinUpstreamCreator)
+
+        @JvmStatic
+        fun data() = listOf(
+            Arguments.of(
+                BlockchainType.ETHEREUM,
+                Runnable { verify(ethereumUpstreamCreator).createUpstream(any(), any()) },
+            ),
+            Arguments.of(
+                BlockchainType.BITCOIN,
+                Runnable { verify(bitcoinUpstreamCreator).createUpstream(any(), any()) },
+            ),
+            Arguments.of(
+                BlockchainType.POLKADOT,
+                Runnable { verify(genericUpstreamCreator).createUpstream(any(), any()) },
+            ),
+            Arguments.of(
+                BlockchainType.STARKNET,
+                Runnable { verify(genericUpstreamCreator).createUpstream(any(), any()) },
+            ),
+            Arguments.of(
+                BlockchainType.UNKNOWN,
+                Runnable { verify(genericUpstreamCreator).createUpstream(any(), any()) },
+            ),
+        )
+    }
+}