multistreams as beans

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

@@ -0,0 +1,27 @@
+package io.emeraldpay.dshackle.config.context
+
+import io.emeraldpay.dshackle.cache.CachesFactory
+import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
+import io.emeraldpay.grpc.BlockchainType
+import io.emeraldpay.grpc.Chain
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+
+@Configuration
+class MultistreamsConfig {
+    @Bean
+    fun allMultistreams(cachesFactory: CachesFactory): List<Multistream> {
+        return Chain.values()
+            .mapNotNull { chain ->
+                when (BlockchainType.from(chain)) {
+                    BlockchainType.EVM_POS -> EthereumPosMultiStream(chain, ArrayList(), cachesFactory.getCaches(chain))
+                    BlockchainType.EVM_POW -> EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain))
+                    BlockchainType.BITCOIN -> BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain))
+                    else -> null
+                }
+            }
+    }
+}

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

@@ -17,39 +17,35 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.cache.CachesEnabled
-import io.emeraldpay.dshackle.cache.CachesFactory
 import io.emeraldpay.dshackle.startup.UpstreamChange
-import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.emeraldpay.grpc.BlockchainType
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
-import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Component
-import org.springframework.stereotype.Repository
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Sinks
-import java.util.Collections
+import java.util.*
-import java.util.concurrent.Callable
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.locks.ReentrantLock
 import javax.annotation.PreDestroy
 import kotlin.concurrent.withLock
 
-@Repository
+@Component
 open class CurrentMultistreamHolder(
-    @Autowired private val cachesFactory: CachesFactory
+    private val multistreams: List<Multistream>
 ) : MultistreamHolder {
 
     private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java)
 
-    private val chainMapping = ConcurrentHashMap<Chain, Multistream>()
+    private val chainMapping = ConcurrentHashMap<Chain, Multistream>().apply {
+        multistreams.forEach { this[it.chain] = it }
+    }
     private val chainsBus = Sinks.many()
         .multicast()
         .directBestEffort<Chain>()
@@ -64,28 +60,22 @@ open class CurrentMultistreamHolder(
                 when (BlockchainType.from(chain)) {
                     BlockchainType.EVM_POW -> {
                         val up = change.upstream.cast(EthereumUpstream::class.java)
-                        val current = chainMapping[chain]
+                        val current = chainMapping.getValue(chain)
-                        val factory = Callable<Multistream> {
+                        processUpdate(change, up, current)
-                            EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain))
-                        }
-                        processUpdate(change, up, current, factory)
                     }
+
                     BlockchainType.EVM_POS -> {
                         val up = change.upstream.cast(EthereumPosUpstream::class.java)
-                        val current = chainMapping[chain]
+                        val current = chainMapping.getValue(chain)
-                        val factory = Callable<Multistream> {
+                        processUpdate(change, up, current)
-                            EthereumPosMultiStream(chain, ArrayList(), cachesFactory.getCaches(chain))
-                        }
-                        processUpdate(change, up, current, factory)
                     }
+
                     BlockchainType.BITCOIN -> {
                         val up = change.upstream.cast(BitcoinUpstream::class.java)
-                        val current = chainMapping[chain]
+                        val current = chainMapping.getValue(chain)
-                        val factory = Callable<Multistream> {
+                        processUpdate(change, up, current)
-                            BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain))
-                        }
-                        processUpdate(change, up, current, factory)
                     }
+
                     else -> {
                         log.error("Update for unsupported chain: $chain")
                     }
@@ -96,27 +86,17 @@ open class CurrentMultistreamHolder(
         }
     }
 
-    fun processUpdate(change: UpstreamChange, up: Upstream, current: Multistream?, factory: Callable<Multistream>) {
+    fun processUpdate(change: UpstreamChange, up: Upstream, current: Multistream) {
         val chain = change.chain
         if (change.type == UpstreamChange.ChangeType.REMOVED) {
-            current?.removeUpstream(up.getId())
+            current.removeUpstream(up.getId())
             log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
         } else {
-            if (current == null) {
+            if (up is CachesEnabled) {
-                val created = factory.call()
+                up.setCaches(current.caches)
-                if (up is CachesEnabled) {
-                    up.setCaches(created.caches)
-                }
-                created.addUpstream(up)
-                created.start()
-                chainMapping[chain] = created
-                chainsBus.tryEmitNext(chain)
-            } else {
-                if (up is CachesEnabled) {
-                    up.setCaches(current.caches)
-                }
-                current.addUpstream(up)
             }
+            current.addUpstream(up)
+
             if (!callTargets.containsKey(chain)) {
                 setupDefaultMethods(chain)
             }
@@ -129,7 +109,10 @@ open class CurrentMultistreamHolder(
     }
 
     override fun getAvailable(): List<Chain> {
-        return Collections.unmodifiableList(chainMapping.keys.toList())
+        return multistreams.asSequence()
+            .filter { it.isAvailable() }
+            .map { it.chain }
+            .toList()
     }
 
     override fun observeChains(): Flux<Chain> {

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

@@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.domain.BlockHash
@@ -90,6 +91,17 @@ class TestingCommons {
         return new CachesFactory(new CacheConfig())
     }
 
+    static List<Multistream> defaultMultistreams() {
+        return [
+                multistreamWithoutUpstreams(Chain.ETHEREUM),
+                multistreamWithoutUpstreams(Chain.ETHEREUM_CLASSIC)
+        ]
+    }
+
+    static Multistream multistreamWithoutUpstreams(Chain chain) {
+        return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain))
+    }
+
     static FileResolver fileResolver() {
         return new FileResolver(new File("src/test/resources"))
     }

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

@@ -26,7 +26,7 @@ class CurrentMultistreamHolderSpec extends Specification {
 
     def "add upstream"() {
         setup:
-        def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
+        def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams())
         def up = new EthereumPosRpcUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api())
         when:
         current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED))
@@ -37,7 +37,7 @@ class CurrentMultistreamHolderSpec extends Specification {
 
     def "add multiple upstreams"() {
         setup:
-        def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
+        def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams())
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
         def up2 = new EthereumRpcUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api())
         def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api())
@@ -53,7 +53,7 @@ class CurrentMultistreamHolderSpec extends Specification {
 
     def "remove upstream"() {
         setup:
-        def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
+        def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams())
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
         def up2 = new EthereumRpcUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api())
         def up3 = new EthereumPosRpcUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api())
@@ -71,7 +71,7 @@ class CurrentMultistreamHolderSpec extends Specification {
 
     def "available after adding"() {
         setup:
-        def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
+        def current = new CurrentMultistreamHolder(TestingCommons.defaultMultistreams())
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
 
         when: