StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: refactoring, more clean class names

Browse Source
This commit is contained in:
Igor Artamonov
2020-05-14 00:12:38 -04:00
parent 98330b8772
commit bbc21684ae
33 changed files with 275 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
View File

@@ -29,15 +29,15 @@ import reactor.core.publisher.Mono
@Service
class Describe(
@Autowired private val upstreams: Upstreams,
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val subscribeStatus: SubscribeStatus
) {
fun describe(requestMono: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
return requestMono.map { _ ->
val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
upstreams.getAvailable().forEach { chain ->
upstreams.getUpstream(chain)?.let { chainUpstreams ->
multistreamHolder.getAvailable().forEach { chain ->
multistreamHolder.getUpstream(chain)?.let { chainUpstreams ->
val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll())
val targets = chainUpstreams.getMethods().getSupportedMethods()
val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder()

10
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -37,7 +37,7 @@ import java.lang.Exception
@Service
open class NativeCall(
@Autowired private val upstreams: Upstreams,
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val objectMapper: ObjectMapper
) {
@@ -88,17 +88,17 @@ open class NativeCall(
return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number)))
}
if (!upstreams.isAvailable(chain)) {
if (!multistreamHolder.isAvailable(chain)) {
return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number)))
}
val upstream = upstreams.getUpstream(chain)
val upstream = multistreamHolder.getUpstream(chain)
?: return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(chain)))
return prepareCall(request, upstream)
}
fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<RawCallDetails>> {
fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: Multistream): Flux<CallContext<RawCallDetails>> {
return request.itemsList.toFlux().map {
val method = it.method
val params = it.payload.toStringUtf8()
@@ -202,7 +202,7 @@ open class NativeCall(
}
open class CallContext<T>(val id: Int,
val upstream: AggregatedUpstream,
val upstream: Multistream,
val matcher: Selector.Matcher,
val callQuorum: CallQuorum,
val payload: T) {

8
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -21,10 +21,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@@ -33,7 +31,7 @@ import reactor.core.publisher.Mono
@Service
class StreamHead(
@Autowired private val upstreams: Upstreams
@Autowired private val multistreamHolder: MultistreamHolder
) {
private val log = LoggerFactory.getLogger(StreamHead::class.java)
@@ -42,7 +40,7 @@ class StreamHead(
return requestMono.map { request ->
Chain.byId(request.type.number)
}.flatMapMany { chain ->
val up = upstreams.getUpstream(chain)
val up = multistreamHolder.getUpstream(chain)
?: return@flatMapMany Flux.error<BlockchainOuterClass.ChainHead>(Exception("Unavailable chain: $chain"))
up.getHead()
.getFlux()

8
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt
View File

@@ -27,13 +27,13 @@ import reactor.core.publisher.Mono
@Service
class SubscribeStatus(
@Autowired private val upstreams: Upstreams
@Autowired private val multistreamHolder: MultistreamHolder
) {
fun subscribeStatus(requestMono: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
return requestMono.flatMapMany {
val ups = upstreams.getAvailable().mapNotNull { chain ->
val chainUpstream = upstreams.getUpstream(chain)
val ups = multistreamHolder.getAvailable().mapNotNull { chain ->
val chainUpstream = multistreamHolder.getUpstream(chain)
chainUpstream?.observeStatus()?.map { avail ->
ChainSubscription(chain, chainUpstream, avail)
}
@@ -60,6 +60,6 @@ class SubscribeStatus(
.build()
}
class ChainSubscription(val chain: Chain, val up: AggregatedUpstream, val avail: UpstreamAvailability)
class ChainSubscription(val chain: Chain, val up: Multistream, val avail: UpstreamAvailability)
}

19
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
View File

@@ -19,9 +19,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
@@ -35,7 +34,7 @@ import kotlin.collections.HashMap
@Service
class TrackBitcoinAddress(
@Autowired private val upstreams: Upstreams
@Autowired private val multistreamHolder: MultistreamHolder
) : TrackAddress {
companion object {
@@ -43,7 +42,7 @@ class TrackBitcoinAddress(
}
override fun isSupported(chain: Chain): Boolean {
return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && upstreams.isAvailable(chain)
return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain)
}
fun allAddresses(request: BlockchainOuterClass.BalanceRequest): List<String>? {
@@ -63,7 +62,7 @@ class TrackBitcoinAddress(
}
}
fun requestBalances(chain: Chain, api: BitcoinChainUpstreams, addresses: List<String>): Flux<AddressBalance> {
fun requestBalances(chain: Chain, api: BitcoinMultistream, addresses: List<String>): Flux<AddressBalance> {
return api.getReader().listUnspent()
.flatMapMany { unspents ->
val result = getTotal(chain, addresses, unspents)
@@ -73,7 +72,7 @@ class TrackBitcoinAddress(
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue)
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)
val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)
?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
if (addresses.isEmpty()) {
@@ -119,9 +118,9 @@ class TrackBitcoinAddress(
override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue)
println("up: ${upstreams.getUpstream(chain)}")
println("up cast: ${upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)}")
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)
println("up: ${multistreamHolder.getUpstream(chain)}")
println("up cast: ${multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)}")
val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)
?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
if (addresses.isEmpty()) {

25
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt
View File

@@ -20,9 +20,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
@@ -38,7 +37,7 @@ import kotlin.math.min
@Service
class TrackBitcoinTx(
@Autowired private val upstreams: Upstreams
@Autowired private val multistreamHolder: MultistreamHolder
) : TrackTx {
companion object {
@@ -46,12 +45,12 @@ class TrackBitcoinTx(
}
override fun isSupported(chain: Chain): Boolean {
return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && upstreams.isAvailable(chain)
return BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain)
}
override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
val chain = Chain.byId(request.chainValue)
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)
val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)
?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
val txid = request.txId
val confirmations = max(min(1, request.confirmationLimit), 12)
@@ -61,7 +60,7 @@ class TrackBitcoinTx(
}.map(this::asProto)
}
fun subscribe(chain: Chain, upstream: BitcoinChainUpstreams, txid: String): Flux<TxStatus> {
fun subscribe(chain: Chain, upstream: BitcoinMultistream, txid: String): Flux<TxStatus> {
return loadExisting(upstream, txid)
.flatMapMany { status ->
if (status.mined) {
@@ -83,7 +82,7 @@ class TrackBitcoinTx(
}
}
fun continueWithMined(upstream: BitcoinChainUpstreams, status: TxStatus): Flux<TxStatus> {
fun continueWithMined(upstream: BitcoinMultistream, status: TxStatus): Flux<TxStatus> {
return upstream.getReader().getBlock(status.blockHash!!)
.map { block ->
TxStatus(status.txid, true, ExtractBlock.getHeight(block), true, status.blockHash, ExtractBlock.getTime(block), ExtractBlock.getDifficulty(block))
@@ -92,7 +91,7 @@ class TrackBitcoinTx(
}
}
fun untilFound(chain: Chain, upstream: BitcoinChainUpstreams, txid: String): Flux<TxStatus> {
fun untilFound(chain: Chain, upstream: BitcoinMultistream, txid: String): Flux<TxStatus> {
return Flux.interval(Duration.ofSeconds(1))
.take(Duration.ofMinutes(10))
.flatMap { loadMempool(upstream, txid) }
@@ -103,20 +102,20 @@ class TrackBitcoinTx(
}
}
fun untilMined(upstream: BitcoinChainUpstreams, tx: TxStatus): Mono<TxStatus> {
fun untilMined(upstream: BitcoinMultistream, tx: TxStatus): Mono<TxStatus> {
return upstream.getHead().getFlux().flatMap {
loadExisting(upstream, tx.txid)
.filter { it.mined }
}.single()
}
fun withConfirmations(upstream: BitcoinChainUpstreams, tx: TxStatus): Flux<TxStatus> {
fun withConfirmations(upstream: BitcoinMultistream, tx: TxStatus): Flux<TxStatus> {
return upstream.getHead().getFlux().map {
tx.withHead(it.height)
}
}
fun loadExisting(api: BitcoinChainUpstreams, txid: String): Mono<TxStatus> {
fun loadExisting(api: BitcoinMultistream, txid: String): Mono<TxStatus> {
val mined = api.getReader().getTx(txid)
return mined.map {
val block = it["blockhash"] as String?
@@ -124,7 +123,7 @@ class TrackBitcoinTx(
}
}
fun loadMempool(upstream: BitcoinChainUpstreams, txid: String): Mono<TxStatus> {
fun loadMempool(upstream: BitcoinMultistream, txid: String): Mono<TxStatus> {
val mempool = upstream.getReader().getMempool().get()
return mempool.map {
if (it.contains(txid)) {

16
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt
View File

@@ -21,8 +21,8 @@ import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.Address
import io.infinitape.etherjar.domain.Wei
@@ -34,13 +34,13 @@ import reactor.core.publisher.Mono
@Service
class TrackEthereumAddress(
@Autowired private val upstreams: Upstreams
@Autowired private val multistreamHolder: MultistreamHolder
) : TrackAddress {
private val log = LoggerFactory.getLogger(TrackEthereumAddress::class.java)
override fun isSupported(chain: Chain): Boolean {
return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain)
return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && multistreamHolder.isAvailable(chain)
}
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
@@ -51,7 +51,7 @@ class TrackEthereumAddress(
override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue)
val head = upstreams.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty()
val head = multistreamHolder.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty()
val balances = initAddress(request)
.flatMap { tracked ->
val current = getBalance(tracked)
@@ -86,14 +86,14 @@ class TrackEthereumAddress(
}
}
fun getUpstream(chain: Chain): EthereumChainUpstream {
return upstreams.getUpstream(chain)?.cast(EthereumChainUpstream::class.java)
fun getUpstream(chain: Chain): EthereumMultistream {
return multistreamHolder.getUpstream(chain)?.cast(EthereumMultistream::class.java)
?: throw SilentException.UnsupportedBlockchain(chain)
}
private fun initAddress(request: BlockchainOuterClass.BalanceRequest): Flux<TrackedAddress> {
val chain = Chain.byId(request.asset.chainValue)
if (!upstreams.isAvailable(chain)) {
if (!multistreamHolder.isAvailable(chain)) {
return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
}
if (request.asset.code?.toLowerCase() != "ether") {

16
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
View File

@@ -23,8 +23,8 @@ import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
@@ -48,7 +48,7 @@ import kotlin.math.min
@Service
class TrackEthereumTx(
@Autowired private val upstreams: Upstreams
@Autowired private val multistreamHolder: MultistreamHolder
) : TrackTx {
companion object {
@@ -63,7 +63,7 @@ class TrackEthereumTx(
private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java)
override fun isSupported(chain: Chain): Boolean {
return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain)
return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && multistreamHolder.isAvailable(chain)
}
override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
@@ -83,12 +83,12 @@ class TrackEthereumTx(
}
fun getUpstream(chain: Chain): EthereumChainUpstream {
return upstreams.getUpstream(chain)?.cast(EthereumChainUpstream::class.java)
fun getUpstream(chain: Chain): EthereumMultistream {
return multistreamHolder.getUpstream(chain)?.cast(EthereumMultistream::class.java)
?: throw SilentException.UnsupportedBlockchain(chain)
}
fun subscribe(base: TxDetails, up: EthereumChainUpstream): Flux<TxDetails> {
fun subscribe(base: TxDetails, up: EthereumMultistream): Flux<TxDetails> {
var latestTx = base
val untilFound = Mono.just(latestTx)
@@ -213,7 +213,7 @@ class TrackEthereumTx(
}
}
fun updateFromBlock(upstream: EthereumChainUpstream, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
fun updateFromBlock(upstream: EthereumMultistream, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) {
val updated = tx.withStatus(
blockHash = blockTx.blockHash,

8
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.CurrentUpstreams
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
@@ -45,7 +45,7 @@ import kotlin.collections.HashMap
@Repository
open class ConfiguredUpstreams(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val currentUpstreams: CurrentUpstreams,
@Autowired private val currentUpstreams: CurrentMultistreamHolder,
@Autowired private val fileResolver: FileResolver,
@Autowired private val config: UpstreamsConfig,
@Autowired private val cachesFactory: CachesFactory
@@ -201,8 +201,7 @@ open class ConfiguredUpstreams(
endpoint.port ?: 2449,
objectMapper,
endpoint.auth,
fileResolver,
cachesFactory
fileResolver
).apply {
timeout = options.timeout
}
@@ -214,7 +213,6 @@ open class ConfiguredUpstreams(
.subscribe(currentUpstreams::update)
}
private fun buildHttpClient(config: UpstreamsConfig.Upstream<out UpstreamsConfig.RpcConnection>): JsonRpcHttpClient? {
val conn = config.connection!!
val urls = ArrayList<URI>()

136
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
View File

@@ -1,136 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Mono
import java.lang.IllegalStateException
import java.time.Duration
/**
* General interface to upstream(s) to a single chain
*/
abstract class ChainUpstreams(
val chain: Chain,
private val upstreams: MutableList<Upstream>,
caches: Caches
) : AggregatedUpstream(caches), Lifecycle {
private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
private var seq = 0
protected var lagObserver: HeadLagObserver? = null
private var subscription: Disposable? = null
open fun init() {
onUpstreamsUpdated()
}
abstract fun updateHead(): Head
abstract fun setHead(head: Head)
override fun getId(): String {
return "!all:${chain.chainCode}"
}
override fun isRunning(): Boolean {
return subscription != null
}
override fun start() {
super.start()
subscription = observeStatus()
.distinctUntilChanged()
.subscribe { printStatus() }
}
override fun stop() {
super.stop()
subscription?.dispose()
subscription = null
getHead().let {
if (it is Lifecycle) {
it.stop()
}
}
lagObserver?.stop()
}
override fun getAll(): List<Upstream> {
return upstreams
}
override fun addUpstream(upstream: Upstream) {
upstreams.add(upstream)
setHead(updateHead())
onUpstreamsUpdated()
}
fun removeUpstream(id: String) {
if (upstreams.removeIf { it.getId() == id }) {
setHead(updateHead())
onUpstreamsUpdated()
}
}
override fun getApiSource(matcher: Selector.Matcher): ApiSource {
val i = seq++
if (seq >= Int.MAX_VALUE / 2) {
seq = 0
}
return FilteredApis(upstreams, matcher, i)
}
override fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
.switchIfEmpty(Mono.error(Exception("No API available")))
}
override fun setLag(lag: Long) {
}
override fun getLag(): Long {
return 0
}
fun printStatus() {
var height: Long? = null
try {
height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.height
} catch (e: IllegalStateException) {
//timout
} catch (e: Exception) {
log.warn("Head processing error: ${e.javaClass} ${e.message}")
}
val statuses = upstreams.map { it.getStatus() }
.groupBy { it }
.map { "${it.key.name}/${it.value.size}" }
.joinToString(",")
val lag = upstreams.map { it.getLag() }
.joinToString(", ")
log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]")
}
}

24
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -21,12 +21,12 @@ import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
@@ -42,14 +42,14 @@ import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
@Repository
class CurrentUpstreams(
class CurrentMultistreamHolder(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val cachesFactory: CachesFactory
): Upstreams {
) : MultistreamHolder {
private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java)
private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java)
private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams>()
private val chainMapping = ConcurrentHashMap<Chain, Multistream>()
private val chainsBus = TopicProcessor.create<Chain>()
private val callTargets = HashMap<Chain, CallMethods>()
private val updateLock = ReentrantLock()
@@ -60,17 +60,17 @@ class CurrentUpstreams(
when (BlockchainType.fromBlockchain(chain)) {
BlockchainType.ETHEREUM -> {
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain] as ChainUpstreams?
val current = chainMapping[chain] as Multistream?
val factory = Callable {
EthereumChainUpstream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams
EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as Multistream
}
processUpdate(change, up, current, factory)
}
BlockchainType.BITCOIN -> {
val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping[chain] as ChainUpstreams?
val current = chainMapping[chain] as Multistream?
val factory = Callable {
BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams
BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as Multistream
}
processUpdate(change, up, current, factory)
}
@@ -81,7 +81,7 @@ class CurrentUpstreams(
}
}
fun processUpdate(change: UpstreamChange, up: Upstream, current: ChainUpstreams?, factory: Callable<ChainUpstreams>) {
fun processUpdate(change: UpstreamChange, up: Upstream, current: Multistream?, factory: Callable<Multistream>) {
val chain = change.chain
if (change.type == UpstreamChange.ChangeType.REMOVED) {
current?.removeUpstream(up.getId())
@@ -109,7 +109,7 @@ class CurrentUpstreams(
}
}
override fun getUpstream(chain: Chain): AggregatedUpstream? {
override fun getUpstream(chain: Chain): Multistream? {
return chainMapping[chain]
}

98
src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -16,7 +16,6 @@
*/
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.*
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
@@ -24,6 +23,8 @@ import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Flux
@@ -38,33 +39,70 @@ import kotlin.concurrent.withLock
/**
* Aggregation of multiple upstreams responding to a single blockchain
*/
abstract class AggregatedUpstream(
abstract class Multistream(
val chain: Chain,
private val upstreams: MutableList<Upstream>,
val caches: Caches
) : Upstream, Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(Multistream::class.java)
}
private var cacheSubscription: Disposable? = null
private val reconfigLock = ReentrantLock()
private var callMethods: CallMethods? = null
private var seq = 0
protected var lagObserver: HeadLagObserver? = null
private var subscription: Disposable? = null
open fun init() {
onUpstreamsUpdated()
}
/**
* Get list of all underlying upstreams
*/
abstract fun getAll(): List<Upstream>
fun getAll(): List<Upstream> {
return upstreams
}
/**
* Add an upstream
*/
abstract fun addUpstream(upstream: Upstream)
fun addUpstream(upstream: Upstream) {
upstreams.add(upstream)
setHead(updateHead())
onUpstreamsUpdated()
}
fun removeUpstream(id: String) {
if (upstreams.removeIf { it.getId() == id }) {
setHead(updateHead())
onUpstreamsUpdated()
}
}
/**
* Get a source for direct APIs
*/
abstract fun getApiSource(matcher: Selector.Matcher): ApiSource
fun getApiSource(matcher: Selector.Matcher): ApiSource {
val i = seq++
if (seq >= Int.MAX_VALUE / 2) {
seq = 0
}
return FilteredApis(upstreams, matcher, i)
}
/**
* Finds an API that executed directly on a remote.
*/
abstract fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
.switchIfEmpty(Mono.error(Exception("No API available")))
}
/**
* Finds an API that leverages caches and other optimizations/transformations of the request.
@@ -109,11 +147,22 @@ abstract class AggregatedUpstream(
}
override fun start() {
subscription = observeStatus()
.distinctUntilChanged()
.subscribe { printStatus() }
}
override fun stop() {
cacheSubscription?.dispose()
cacheSubscription = null
subscription?.dispose()
subscription = null
getHead().let {
if (it is Lifecycle) {
it.stop()
}
}
lagObserver?.stop()
}
fun onHeadUpdated(head: Head) {
@@ -125,6 +174,43 @@ abstract class AggregatedUpstream(
}
}
abstract fun updateHead(): Head
abstract fun setHead(head: Head)
override fun getId(): String {
return "!all:${chain.chainCode}"
}
override fun isRunning(): Boolean {
return subscription != null
}
override fun setLag(lag: Long) {
}
override fun getLag(): Long {
return 0
}
fun printStatus() {
var height: Long? = null
try {
height = getHead().getFlux().next().block(Duration.ofSeconds(1))?.height
} catch (e: java.lang.IllegalStateException) {
//timout
} catch (e: Exception) {
log.warn("Head processing error: ${e.javaClass} ${e.message}")
}
val statuses = upstreams.map { it.getStatus() }
.groupBy { it }
.map { "${it.key.name}/${it.value.size}" }
.joinToString(",")
val lag = upstreams.map { it.getLag() }
.joinToString(", ")
log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses, lag=[$lag]")
}
// --------------------------------------------------------------------------------------------------------
class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/MultistreamHolder.kt
View File

@@ -16,16 +16,15 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
interface Upstreams {
fun getUpstream(chain: Chain): AggregatedUpstream?
/**
* Holds Multistreams configured for a chain.
*/
interface MultistreamHolder {
fun getUpstream(chain: Chain): Multistream?
fun getAvailable(): List<Chain>
fun observeChains(): Flux<Chain>
fun getDefaultMethods(chain: Chain): CallMethods

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -27,20 +27,18 @@ import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono
open class BitcoinChainUpstreams(
open class BitcoinMultistream(
chain: Chain,
val upstreams: MutableList<BitcoinUpstream>,
caches: Caches,
private val objectMapper: ObjectMapper
) : ChainUpstreams(chain, upstreams as MutableList<Upstream>, caches), Lifecycle {
) : Multistream(chain, upstreams as MutableList<Upstream>, caches), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinChainUpstreams::class.java)
private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)
}
private var head: Head? = null
//TODO head
private var reader = BitcoinReader(this, EmptyHead(), objectMapper)
override fun init() {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
View File

@@ -26,7 +26,7 @@ import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.cast
open class BitcoinReader(
private val upstreams: BitcoinChainUpstreams,
private val upstreams: BitcoinMultistream,
head: Head,
private val objectMapper: ObjectMapper
) : Lifecycle {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt
View File

@@ -30,7 +30,7 @@ import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock
open class CachingMempoolData(
private val upstreams: BitcoinChainUpstreams,
private val upstreams: BitcoinMultistream,
private val head: Head,
private val objectMapper: ObjectMapper
) : Lifecycle {

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstream.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -28,15 +28,15 @@ import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono
open class EthereumChainUpstream(
open class EthereumMultistream(
chain: Chain,
val upstreams: MutableList<EthereumUpstream>,
caches: Caches,
private val objectMapper: ObjectMapper
) : ChainUpstreams(chain, upstreams as MutableList<Upstream>, caches) {
) : Multistream(chain, upstreams as MutableList<Upstream>, caches) {
companion object {
private val log = LoggerFactory.getLogger(EthereumChainUpstream::class.java)
private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)
}
private var head: Head? = null

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt
View File

@@ -21,10 +21,8 @@ import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.data.*
import io.emeraldpay.dshackle.reader.*
import io.emeraldpay.dshackle.upstream.AggregatedUpstream
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.domain.Address
@@ -35,7 +33,6 @@ import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.BlockTag
import io.infinitape.etherjar.rpc.json.TransactionJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
@@ -48,7 +45,7 @@ import java.util.concurrent.TimeoutException
import java.util.function.Function
open class EthereumReader(
private val up: AggregatedUpstream,
private val up: Multistream,
private val caches: Caches,
private val objectMapper: ObjectMapper
) : Lifecycle {

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -22,8 +22,6 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
@@ -39,7 +37,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -22,7 +22,6 @@ import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.startup.UpstreamChange
@@ -30,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
import io.emeraldpay.grpc.Chain
import io.grpc.ManagedChannelBuilder
import io.grpc.netty.NettyChannelBuilder
import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient
import io.netty.handler.ssl.*
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.exception.ExceptionUtils
@@ -50,8 +48,7 @@ class GrpcUpstreams(
private val port: Int,
private val objectMapper: ObjectMapper,
private val auth: AuthConfig.ClientTlsAuth? = null,
private val fileResolver: FileResolver,
private val cachesFactory: CachesFactory
private val fileResolver: FileResolver
) {
private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)