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

solution: refactoring

Browse Source
This commit is contained in:
Igor Artamonov
2020-02-24 22:14:24 -05:00
parent b255bed3a1
commit 2816b5b8fc
30 changed files with 158 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
@@ -43,13 +44,13 @@ class Describe(
.setStatus(status)
chainUpstreams.getAll().let { ups ->
ups.forEach { up ->
val nodes = NodeDetailsList()
val nodes = QuorumForLabels()
if (up is EthereumUpstream) {
nodes.add(up.node)
} else if (up is GrpcUpstream) {
nodes.add(up.getNodes())
}
nodes.getNodes().forEach { node ->
nodes.getAll().forEach { node ->
val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder()
.setQuorum(node.quorum)
.addAllLabels(node.labels.entries.map { label ->

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt → src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -13,12 +13,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.startup
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfigReader
import io.emeraldpay.dshackle.upstream.CurrentUpstreams
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs
@@ -172,7 +174,7 @@ open class ConfiguredUpstreams(
val ethereumUpstream = EthereumUpstream(
config.id!!,
chain, rpcApi!!, wsApi, options,
NodeDetailsList.NodeDetails(1, config.labels),
QuorumForLabels.QuorumItem(1, config.labels),
methods)
ethereumUpstream.start()
currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED))

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/NodeDetailsList.kt → src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt
View File

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.startup
import io.emeraldpay.dshackle.config.UpstreamsConfig
import java.util.*
@@ -22,17 +22,20 @@ import kotlin.collections.ArrayList
import kotlin.concurrent.read
import kotlin.concurrent.write
class NodeDetailsList {
/**
* Summary details over few upstream nodes. Provides aggregate quorum for nodes with particular label
*/
class QuorumForLabels {
private val lock = ReentrantReadWriteLock()
private val nodes = ArrayList<NodeDetails>()
private val nodes = ArrayList<QuorumItem>()
fun add(node: NodeDetails) {
fun add(node: QuorumItem) {
lock.read {
val existing = nodes.find { it.labels == node.labels }
lock.write {
if (existing != null) {
val merged = NodeDetails(existing.quorum + node.quorum, existing.labels)
val merged = QuorumItem(existing.quorum + node.quorum, existing.labels)
nodes.remove(existing)
nodes.add(merged)
} else {
@@ -42,15 +45,18 @@ class NodeDetailsList {
}
}
fun add(nodes: NodeDetailsList) {
fun add(nodes: QuorumForLabels) {
nodes.nodes.forEach { node -> this.add(node) }
}
fun getNodes(): List<NodeDetails> {
fun getAll(): List<QuorumItem> {
return Collections.unmodifiableList(nodes)
}
class NodeDetails(val quorum: Int, val labels: UpstreamsConfig.Labels) {
/**
* Details for a single element (upstream, node or aggregation)
*/
class QuorumItem(val quorum: Int, val labels: UpstreamsConfig.Labels) {
companion object {
}
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt → src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt
View File

@@ -13,10 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.startup
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.grpc.Chain
class UpstreamChange(

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

@@ -18,6 +18,8 @@ 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.upstream.calls.AggregatedCallMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import org.springframework.context.Lifecycle
import reactor.core.Disposable

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

@@ -18,6 +18,9 @@ package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import reactor.core.publisher.Flux

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt
View File

@@ -15,6 +15,7 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux

20
src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedCallMethods.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethods.kt
View File

@@ -13,12 +13,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.upstream.calls
import io.emeraldpay.dshackle.quorum.CallQuorum
import java.util.*
import kotlin.collections.HashSet
/**
* Aggregation over several parent configuration. It dispatches call to a first delegate that supports it.
*/
class AggregatedCallMethods(
private val delegates: Collection<CallMethods>
): CallMethods {
@@ -31,24 +34,39 @@ class AggregatedCallMethods(
allMethods = Collections.unmodifiableSet(buf)
}
/**
* Finds first delegate that has Allowed that method and returns its Quorum
*/
override fun getQuorumFor(method: String): CallQuorum {
return delegates.find {
it.isAllowed(method)
}?.getQuorumFor(method) ?: throw IllegalStateException("No quorum for $method")
}
/**
* Checks if ANY of delegates supports the method
*/
override fun isAllowed(method: String): Boolean {
return delegates.any { it.isAllowed(method) }
}
/**
* Returns all available methods, accessible through at least one of delegates
*/
override fun getSupportedMethods(): Set<String> {
return allMethods
}
/**
* @return true if there is at least one delegate that allows the method and it's hardcoded on that delegate
*/
override fun isHardcoded(method: String): Boolean {
return delegates.any { it.isAllowed(method) && it.isHardcoded(method) }
}
/**
* Executed the method on the first delegate that supports it as a hardcoded method
*/
override fun executeHardcoded(method: String): Any {
return delegates.find {
it.isAllowed(method) && it.isHardcoded(method)

25
src/main/kotlin/io/emeraldpay/dshackle/upstream/CallMethods.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallMethods.kt
View File

@@ -13,14 +13,37 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.upstream.calls
import io.emeraldpay.dshackle.quorum.CallQuorum
/**
* Configuration for upstream calls
*/
interface CallMethods {
/**
* @return CallQuorum configured for the specified method
*/
fun getQuorumFor(method: String): CallQuorum
/**
* @return false is call for that method is not allowed. Allowed method may be also Hardcoded
*/
fun isAllowed(method: String): Boolean
/**
* @return list of all allowed methods.
*/
fun getSupportedMethods(): Set<String>
/**
* @return true if the method should not be executed on upstream, but accessed through this class
*/
fun isHardcoded(method: String): Boolean
/**
* Read [supposed to be predefined] method from this config
*/
fun executeHardcoded(method: String): Any
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/DirectCallMethods.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt
View File

@@ -13,12 +13,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.upstream.calls
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum
import java.util.*
/**
* Configuration that uses [AlwaysQuorum] for all available methods. The methods list itself
* is provided in constructor (or empty otherwise)
*/
class DirectCallMethods(private val methods: Set<String>) : CallMethods {
constructor(): this(emptySet())

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ManagedCallMethods.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt
View File

@@ -13,12 +13,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.upstream.calls
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum
import java.util.*
/**
* Wrapper on top of another configuration, that may disable or enable additional methods on top of it.
* If a new method enabled, then its Quorum will be [AlwaysQuorum]. For other methods it delegates all to the provided
* parent config.
*/
class ManagedCallMethods(
private val delegate: CallMethods,
private val enabled: Set<String>,

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/QuorumBasedMethods.kt
View File

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
package io.emeraldpay.dshackle.upstream.calls
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.quorum.*
@@ -22,6 +22,10 @@ import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcException
import java.util.*
/**
* Default configuration for Ethereum based RPC. Defines optimal Quorum strategies for different methods, and provides
* hardcoded results for base methods, such as `net_version`, `web3_clientVersion` and similar
*/
class QuorumBasedMethods(
private val objectMapper: ObjectMapper,
private val chain: Chain

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
View File

@@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.upstream.CallMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.grpc.Status
import io.grpc.StatusRuntimeException
import io.infinitape.etherjar.domain.BlockHash

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

@@ -18,7 +18,10 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
@@ -32,12 +35,12 @@ open class EthereumUpstream(
private val api: DirectEthereumApi,
private val ethereumWs: EthereumWs? = null,
private val options: UpstreamsConfig.Options,
val node: NodeDetailsList.NodeDetails,
val node: QuorumForLabels.QuorumItem,
private val targets: CallMethods
): DefaultUpstream(), CachesEnabled, Lifecycle {
constructor(id: String, chain: Chain, api: DirectEthereumApi): this(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()),
UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
DirectCallMethods())

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
View File

@@ -24,13 +24,15 @@ 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.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient
import io.infinitape.etherjar.rpc.json.BlockJson
@@ -62,7 +64,7 @@ open class GrpcUpstream(
private var caches: Caches? = null
private val options = UpstreamsConfig.Options.getDefaults()
private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
private val nodes = AtomicReference<QuorumForLabels>(QuorumForLabels())
private val head = DefaultEthereumHead()
private var targets: CallMethods? = null
private var headSubscription: Disposable? = null
@@ -148,10 +150,10 @@ open class GrpcUpstream(
fun init(conf: BlockchainOuterClass.DescribeChain) {
targets = DirectCallMethods(conf.supportedMethodsList.toSet())
val nodes = NodeDetailsList()
val nodes = QuorumForLabels()
val allLabels = ArrayList<UpstreamsConfig.Labels>()
conf.nodesList.forEach { remoteNode ->
val node = NodeDetailsList.NodeDetails(remoteNode.quorum,
val node = QuorumForLabels.QuorumItem(remoteNode.quorum,
remoteNode.labelsList.let { provided ->
val labels = UpstreamsConfig.Labels()
provided.forEach {
@@ -176,7 +178,7 @@ open class GrpcUpstream(
)
}
fun getNodes(): NodeDetailsList {
fun getNodes(): QuorumForLabels {
return nodes.get()
}
@@ -191,7 +193,7 @@ open class GrpcUpstream(
}
override fun isAvailable(): Boolean {
return getStatus() == UpstreamAvailability.OK && head.getCurrent() != null && nodes.get().getNodes().any {
return getStatus() == UpstreamAvailability.OK && head.getCurrent() != null && nodes.get().getAll().any {
it.quorum > 0
}
}

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

@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.UpstreamChange
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.grpc.Chain
import io.grpc.ManagedChannelBuilder
import io.grpc.netty.NettyChannelBuilder
@@ -36,7 +36,6 @@ import reactor.core.publisher.Flux
import java.net.ConnectException
import java.time.Duration
import java.util.*
import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock