From 443839062f8306ee9d8b210fe3f4572a91facc9a Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Tue, 15 Nov 2022 20:52:15 +0400 Subject: [PATCH] multistreams as beans --- .../config/context/MultistreamsConfig.kt | 27 ++++++++ .../upstream/CurrentMultistreamHolder.kt | 69 +++++++------------ .../dshackle/test/TestingCommons.groovy | 12 ++++ .../CurrentMultistreamHolderSpec.groovy | 8 +-- 4 files changed, 69 insertions(+), 47 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt new file mode 100644 index 00000000..7ffdd93d --- /dev/null +++ b/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 { + 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 + } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index 7587d6c2..2de935f6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/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.Repository +import org.springframework.stereotype.Component import reactor.core.publisher.Flux import reactor.core.publisher.Sinks -import java.util.Collections -import java.util.concurrent.Callable +import java.util.* 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 ) : MultistreamHolder { private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java) - private val chainMapping = ConcurrentHashMap() + private val chainMapping = ConcurrentHashMap().apply { + multistreams.forEach { this[it.chain] = it } + } private val chainsBus = Sinks.many() .multicast() .directBestEffort() @@ -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 factory = Callable { - EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) - } - processUpdate(change, up, current, factory) + val current = chainMapping.getValue(chain) + processUpdate(change, up, current) } + BlockchainType.EVM_POS -> { val up = change.upstream.cast(EthereumPosUpstream::class.java) - val current = chainMapping[chain] - val factory = Callable { - EthereumPosMultiStream(chain, ArrayList(), cachesFactory.getCaches(chain)) - } - processUpdate(change, up, current, factory) + val current = chainMapping.getValue(chain) + processUpdate(change, up, current) } + BlockchainType.BITCOIN -> { val up = change.upstream.cast(BitcoinUpstream::class.java) - val current = chainMapping[chain] - val factory = Callable { - BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) - } - processUpdate(change, up, current, factory) + val current = chainMapping.getValue(chain) + processUpdate(change, up, current) } + 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) { + 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) { - val created = factory.call() - 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) + if (up is CachesEnabled) { + up.setCaches(current.caches) } + current.addUpstream(up) + if (!callTargets.containsKey(chain)) { setupDefaultMethods(chain) } @@ -129,7 +109,10 @@ open class CurrentMultistreamHolder( } override fun getAvailable(): List { - return Collections.unmodifiableList(chainMapping.keys.toList()) + return multistreams.asSequence() + .filter { it.isAvailable() } + .map { it.chain } + .toList() } override fun observeChains(): Flux { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index fbec4d77..6ad08167 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/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 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")) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy index 2f38eecd..0ec23584 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolderSpec.groovy +++ b/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: