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

solution: connects to remote dshackle

Browse Source
This commit is contained in:
Igor Artamonov
2019-07-25 22:53:02 -04:00
parent 499a92e4d6
commit fd8c35bec2
31 changed files with 1320 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -2,4 +2,5 @@
build/
out/
*.iml
*.yaml
*.yaml
testsetup/

5
build.gradle
View File

@@ -44,7 +44,7 @@ configurations {
}
dependencies {
compile "io.emeraldpay:emerald-grpc:0.4"
compile "io.emeraldpay:emerald-grpc:0.6-SNAPSHOT"
compile "io.grpc:grpc-protobuf:${grpcVersion}"
compile "io.grpc:grpc-stub:${grpcVersion}"
@@ -63,6 +63,7 @@ dependencies {
compile 'io.projectreactor:reactor-core:3.2.9.RELEASE'
compile 'io.projectreactor.addons:reactor-extra:3.2.3.RELEASE'
compile 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.0.M1'
compile 'com.salesforce.servicelibs:reactor-grpc:0.10.0'
compile 'org.yaml:snakeyaml:1.24'
@@ -93,7 +94,9 @@ dependencies {
testCompile "org.codehaus.groovy:groovy:$groovyVersion"
testCompile 'cglib:cglib-nodep:3.2.12'
testCompile "org.spockframework:spock-core:$spockVersion"
testCompile "io.grpc:grpc-testing:${grpcVersion}"
}
compileKotlin {

111
src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java
View File

@@ -1,7 +1,7 @@
package io.emeraldpay.dshackle.config;
import org.yaml.snakeyaml.TypeDescription;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.introspector.GenericProperty;
import org.yaml.snakeyaml.introspector.MethodProperty;
import org.yaml.snakeyaml.introspector.Property;
import org.yaml.snakeyaml.introspector.PropertySubstitute;
@@ -12,6 +12,10 @@ import org.yaml.snakeyaml.nodes.ScalarNode;
import javax.annotation.Nullable;
import java.beans.IntrospectionException;
import java.beans.PropertyDescriptor;
import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Type;
import java.net.URI;
import java.util.Arrays;
import java.util.Collections;
@@ -201,6 +205,8 @@ public class UpstreamsConfig {
public static class Endpoint {
private EndpointType type;
private URI url;
private String host;
private int port;
@Nullable
private Auth auth;
private Boolean enabled = true;
@@ -248,6 +254,22 @@ public class UpstreamsConfig {
public void setOrigin(@Nullable URI origin) {
this.origin = origin;
}
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
}
public static class Auth {
@@ -262,7 +284,12 @@ public class UpstreamsConfig {
}
}
public static class BasicAuth extends Auth {
public static interface WithKey {
public String getKey();
public void setKey(String key);
}
public static class BasicAuth extends Auth implements WithKey {
private String key;
public String getKey() {
@@ -274,18 +301,72 @@ public class UpstreamsConfig {
}
}
public static class TlsAuth extends Auth implements WithKey {
private String ca;
private String certificate;
private String key;
public String getCa() {
return ca;
}
public void setCa(String ca) {
this.ca = ca;
}
public String getCertificate() {
return certificate;
}
public void setCertificate(String certificate) {
this.certificate = certificate;
}
public String getKey() {
return key;
}
public void setKey(String key) {
this.key = key;
}
}
public static class AuthYaml extends TypeDescription {
public AuthYaml() {
super(Auth.class);
}
public Class getImpl() {
try {
Field f = TypeDescription.class.getDeclaredField("impl");
f.setAccessible(true);
return (Class) f.get(this);
} catch (NoSuchFieldException | IllegalAccessException e) {
e.printStackTrace();
}
return null;
}
@Override
public Property getProperty(String name) {
if ("key".equals(name)) {
try {
return new MethodProperty(new PropertyDescriptor("key", BasicAuth.class, "getKey", "setKey"));
return new MethodProperty(new PropertyDescriptor("key", WithKey.class, "getKey", "setKey"));
} catch (IntrospectionException e) {
e.printStackTrace();
}
}
if ("ca".equals(name)) {
try {
return new MethodProperty(new PropertyDescriptor("ca", TlsAuth.class, "getCa", "setCa"));
} catch (IntrospectionException e) {
e.printStackTrace();
}
}
if ("certificate".equals(name)) {
try {
return new MethodProperty(new PropertyDescriptor("certificate", TlsAuth.class, "getCertificate", "setCertificate"));
} catch (IntrospectionException e) {
e.printStackTrace();
}
@@ -293,22 +374,28 @@ public class UpstreamsConfig {
return super.getProperty(name);
}
private Optional<ScalarNode> getValue(MappingNode mappingNode, String key) {
return mappingNode.getValue()
.stream()
.filter((n) -> n.getKeyNode() instanceof ScalarNode && n.getValueNode() instanceof ScalarNode)
.filter((n) -> {
ScalarNode sn = (ScalarNode)n.getKeyNode();
return "type".equals(sn.getValue());
})
.map((n) -> (ScalarNode)n.getValueNode())
.findFirst();
}
@Override
public Object newInstance(Node node) {
if (node instanceof MappingNode) {
MappingNode mappingNode = (MappingNode)node;
Optional<ScalarNode> type = mappingNode.getValue()
.stream()
.filter((n) -> n.getKeyNode() instanceof ScalarNode && n.getValueNode() instanceof ScalarNode)
.filter((n) -> {
ScalarNode sn = (ScalarNode)n.getKeyNode();
return "type".equals(sn.getValue());
})
.map((n) -> (ScalarNode)n.getValueNode())
.findFirst();
Optional<ScalarNode> type = getValue(mappingNode, "type");
if (type.isPresent()) {
if ("basic".equals(type.get().getValue())) {
return new BasicAuth();
} else if ("tls".equals(type.get().getValue())) {
return new TlsAuth();
} else {
throw new IllegalArgumentException("Unsupported auth type: " + type.get().getValue());
}

12
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -16,7 +16,9 @@ class BlockchainRpc(
@Autowired private val nativeCall: NativeCall,
@Autowired private val streamHead: StreamHead,
@Autowired private val trackTx: TrackTx,
@Autowired private val trackAddress: TrackAddress
@Autowired private val trackAddress: TrackAddress,
@Autowired private val describe: Describe,
@Autowired private val subscribeStatus: SubscribeStatus
): BlockchainGrpc.BlockchainImplBase() {
private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)
@@ -56,4 +58,12 @@ class BlockchainRpc(
}
}
override fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver<BlockchainOuterClass.DescribeResponse>) {
describe.describe(request, responseObserver)
}
override fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.ChainStatus>) {
subscribeStatus.subscribeStatus(request, responseObserver)
}
}

43
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt Normal file
View File

@@ -0,0 +1,43 @@
package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
import io.emeraldpay.dshackle.upstream.Upstreams
import io.grpc.stub.StreamObserver
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@Service
class Describe(
@Autowired private val upstreams: Upstreams
) {
fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver<BlockchainOuterClass.DescribeResponse>) {
val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
upstreams.getAvailable().forEach { chain ->
upstreams.ethereumUpstream(chain).let { chainUpstreams ->
val quorum = chainUpstreams.getAll().map { u ->
if (u.getStatus() == UpstreamAvailability.OK) {
u.getOptions().quorum
} else {
0
}
}.sum()
val available = chainUpstreams.getAll().any { u ->
u.getStatus() == UpstreamAvailability.OK
}
resp.addChains(
BlockchainOuterClass.DescribeChain.newBuilder()
.setChain(Common.ChainRef.forNumber(chain.id))
.setQuorum(quorum)
.setAvailable(available)
.build()
)
}
}
responseObserver.onNext(resp.build())
responseObserver.onCompleted()
}
}

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

@@ -3,6 +3,7 @@ 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.upstream.ConfiguredUpstreams
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.grpc.Chain
import io.grpc.stub.StreamObserver
@@ -27,7 +28,7 @@ class NativeCall(
if (chain == Chain.UNSPECIFIED) {
throw Exception("Invalid chain id: ${request.chain.number}")
}
val upstream = upstreams.ethereumUpstream(chain)?.api ?: throw Exception("Chain ${chain.id} is unavailable")
val upstream = upstreams.ethereumUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable")
request.itemsList.toFlux()
.map {
val method = it.target

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

@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.rpc
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.grpc.Chain
import io.grpc.stub.StreamObserver
@@ -12,7 +13,6 @@ import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.toFlux
import java.lang.Exception
import java.util.*
import java.util.concurrent.ConcurrentLinkedQueue
import javax.annotation.PostConstruct
import kotlin.collections.HashMap
@@ -28,7 +28,7 @@ class StreamHead(
@PostConstruct
fun init() {
listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_MORDEN, Chain.TESTNET_KOVAN).forEach { chain ->
if (upstreams.ethereumUpstream(chain)?.head != null) {
if (upstreams.ethereumUpstream(chain)?.getHead() != null) {
clients[chain] = ConcurrentLinkedQueue()
subscribe(chain)
}
@@ -36,7 +36,7 @@ class StreamHead(
}
private fun subscribe(chain: Chain) {
upstreams.ethereumUpstream(chain)!!.head.getFlux()
upstreams.ethereumUpstream(chain)!!.getHead().getFlux()
.doOnComplete {
log.info("Closing streams for ${chain.chainCode}")
clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client ->
@@ -68,7 +68,7 @@ class StreamHead(
fun process(chain: Chain, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
val upstream = upstreams.ethereumUpstream(chain) ?: return false
val head = upstream.head.getHead()
val head = upstream.getHead().getHead()
return head.map {
notify(chain, it, client)
}.defaultIfEmpty(false).block()!!

19
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt Normal file
View File

@@ -0,0 +1,19 @@
package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.upstream.Upstreams
import io.grpc.stub.StreamObserver
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@Service
class SubscribeStatus(
@Autowired private val upstreams: Upstreams
) {
fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver<BlockchainOuterClass.ChainStatus>) {
upstreams.getAvailable().forEach { chain ->
}
}
}

30
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt
View File

@@ -2,12 +2,12 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.grpc.Chain
import io.grpc.stub.StreamObserver
import io.infinitape.etherjar.domain.Address
import io.infinitape.etherjar.domain.Wei
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockTag
import org.springframework.beans.factory.annotation.Autowired
@@ -17,14 +17,11 @@ import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.toFlux
import reactor.math.sum
import reactor.util.function.Tuple2
import reactor.util.function.Tuples
import java.lang.Exception
import java.time.Duration
import java.time.Instant
import java.util.*
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.Future
import javax.annotation.PostConstruct
@Service
@@ -40,7 +37,7 @@ class TrackAddress(
fun init() {
allChains.forEach { chain ->
clients[chain] = ConcurrentLinkedQueue()
upstreams.ethereumUpstream(chain)?.head?.let { head ->
upstreams.ethereumUpstream(chain)?.getHead()?.let { head ->
head.getFlux().subscribe { verifyAll(chain) }
}
}
@@ -120,25 +117,18 @@ class TrackAddress(
}
private fun verify(chain: Chain, group: List<TrackedAddress>): Flux<TrackedAddress> {
val up = upstreams.ethereumUpstream(chain)!!
val up = upstreams.ethereumUpstream(chain)
return group.toFlux()
.reduce<Tuple2<Batch, ArrayList<Update>>>(Tuples.of(Batch(), ArrayList())) { batch, a ->
val f = batch.t1.add(Commands.eth().getBalance(a.address, BlockTag.LATEST));
batch.t2.add(Update(a, f))
batch
}
.flatMap {
Mono.fromCompletionStage(up.api.execute(it.t1))
.thenReturn(it.t2)
}
.flatMapMany {
it.toFlux()
.flatMap { a ->
up.getApi()
.executeAndConvert(Commands.eth().getBalance(a.address, BlockTag.LATEST))
.map { Update(a, it) }
}
.filter {
it.addr.balance == null || it.addr.balance != it.value.get()
it.addr.balance == null || it.addr.balance != it.value
}
.doOnNext {
it.addr.balance = it.value.get()
it.addr.balance = it.value
}
.map {
it.addr
@@ -163,7 +153,7 @@ class TrackAddress(
return sent
}
class Update(val addr: TrackedAddress, val value: Future<Wei>)
class Update(val addr: TrackedAddress, val value: Wei)
class TrackedAddress(val chain: Chain,
val stream: StreamSender<BlockchainOuterClass.AddressBalance>,

24
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
View File

@@ -3,19 +3,17 @@ package io.emeraldpay.dshackle.rpc
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.grpc.Chain
import io.grpc.stub.StreamObserver
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.Commands
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono
import reactor.core.publisher.toFlux
import reactor.kotlin.core.publisher.switchIfEmpty
import java.lang.Exception
import java.math.BigInteger
import java.time.Duration
@@ -37,7 +35,7 @@ class TrackTx(
fun init() {
listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain ->
clients[chain] = ConcurrentLinkedQueue()
upstreams.ethereumUpstream(chain)?.head?.let { head ->
upstreams.ethereumUpstream(chain)?.getHead()?.let { head ->
head.getFlux().subscribe { verifyAll(chain) }
}
}
@@ -63,9 +61,9 @@ class TrackTx(
}
private fun loadWeight(tx: TrackedTx): Mono<TrackedTx> {
val batch = Batch()
val execution = Mono
.fromCompletionStage(batch.add(Commands.eth().getBlock(tx.status.blockHash)))
val upstream = upstreams.ethereumUpstream(tx.chain)
return upstream.getApi()
.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash))
.map { block ->
if (block != null && block.number != null && block.totalDifficulty != null) {
tx.withStatus(
@@ -78,18 +76,14 @@ class TrackTx(
)
}
}
val upstream = upstreams.ethereumUpstream(tx.chain)!!
upstream.api.execute(batch)
return execution
}
private fun verify(tx: TrackedTx): Boolean {
val found = tx.status.found
val mined = tx.status.mined
val batch = Batch()
val execution = Mono.fromCompletionStage(batch.add(Commands.eth().getTransaction(tx.txid)))
val upstream = upstreams.ethereumUpstream(tx.chain)!!
upstream.api.execute(batch)
val upstream = upstreams.ethereumUpstream(tx.chain)
val execution = upstream.getApi()
.executeAndConvert(Commands.eth().getTransaction(tx.txid))
val update = execution.flatMap {
if (it.blockNumber != null
&& it.blockHash != null && it.blockHash != ZERO_BLOCK) {
@@ -100,7 +94,7 @@ class TrackTx(
mined = true,
confirmation = 1
)
return@flatMap upstream.head.getHead().map { head ->
return@flatMap upstream.getHead().getHead().map { head ->
tx.withStatus(
confirmation = head.number - tx.status.height!! + 1
)

52
src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt Normal file
View File

@@ -0,0 +1,52 @@
package io.emeraldpay.dshackle.upstream
abstract class AggregatedUpstreams {
abstract fun getAll(): List<Upstream>
abstract fun addUpstream(upstream: Upstream)
abstract fun getApis(quorum: Int): Iterator<EthereumApi>
abstract fun getApi(): EthereumApi
abstract fun getHead(): EthereumHead
class SingleApi(
private val quorumApi: QuorumApi
): Iterator<EthereumApi> {
private var consumed = false
override fun hasNext(): Boolean {
return !consumed && quorumApi.hasNext()
}
override fun next(): EthereumApi {
consumed = true
return quorumApi.next()
}
}
class QuorumApi(
private val apis: List<Upstream>,
private val quorum: Int,
private var pos: Int
): Iterator<EthereumApi> {
private var consumed = 0
override fun hasNext(): Boolean {
return consumed < quorum
}
override fun next(): EthereumApi {
val start = pos
while (pos < start + apis.size) {
val api = apis[pos++ % apis.size]
if (api.isAvailable()) {
consumed++
return api.getApi()
}
}
throw IllegalStateException("No upstream API available")
}
}
}

93
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt
View File

@@ -1,93 +0,0 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import org.slf4j.LoggerFactory
import java.lang.IllegalStateException
import java.time.Duration
class ChainConnect(
val chain: Chain,
val upstreams: List<Upstream>
) {
private val log = LoggerFactory.getLogger(ChainConnect::class.java)
private var seq = 0
val head: EthereumHead = if (upstreams.size == 1) {
upstreams.first().head
} else {
EthereumHeadMerge(upstreams.map { it.head })
}
val api: EthereumApi
get() {
return getApis(1).next()
}
fun getApis(quorum: Int): Iterator<EthereumApi> {
val i = seq++
if (seq >= Int.MAX_VALUE / 2) {
seq = 0
}
return QuorumApi(upstreams, 1, seq)
}
fun printStatus() {
var height: Long = -1
try {
height = head.getHead().block(Duration.ofSeconds(1))?.number ?: -1
} catch (e: Exception) { }
val statuses = upstreams.map { it.getStatus() }
.groupBy { it }
.map { "${it.key.name}/${it.value.size}" }
.joinToString(",")
log.info("State of ${chain.chainCode}: height=$height, status=$statuses")
}
class SingleApi(
private val quorumApi: QuorumApi
): Iterator<EthereumApi> {
private var consumed = false
override fun hasNext(): Boolean {
return !consumed && quorumApi.hasNext()
}
override fun next(): EthereumApi {
consumed = true
return quorumApi.next()
}
}
class QuorumApi(
private val apis: List<Upstream>,
private val quorum: Int,
private var pos: Int
): Iterator<EthereumApi> {
private var consumed = 0
override fun hasNext(): Boolean {
return consumed < quorum
}
override fun next(): EthereumApi {
val start = pos
while (pos < start + apis.size) {
val api = apis[pos++ % apis.size]
if (api.isAvailable()) {
consumed++
return api.api
}
}
throw IllegalStateException("No upstream API available")
}
}
}

72
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt Normal file
View File

@@ -0,0 +1,72 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import java.lang.IllegalStateException
import java.time.Duration
class ChainUpstreams (
val chain: Chain,
private val upstreams: MutableList<Upstream>
) : AggregatedUpstreams() {
private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
private var seq = 0
private var head: EthereumHead
init {
head = updateHead()
}
internal fun updateHead(): EthereumHead {
return if (upstreams.size == 1) {
upstreams.first().getHead()
} else {
EthereumHeadMerge(upstreams.map { it.getHead() })
}
}
override fun getAll(): List<Upstream> {
return upstreams
}
override fun addUpstream(upstream: Upstream) {
upstreams.add(upstream)
head = updateHead()
}
override fun getApis(quorum: Int): Iterator<EthereumApi> {
val i = seq++
if (seq >= Int.MAX_VALUE / 2) {
seq = 0
}
return QuorumApi(upstreams, 1, seq)
}
override fun getApi(): EthereumApi {
return getApis(1).next()
}
override fun getHead(): EthereumHead {
return head
}
fun printStatus() {
var height: Long = -1
try {
height = head.getHead().block(Duration.ofSeconds(1))?.number ?: -1
} 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(",")
log.info("State of ${chain.chainCode}: height=$height, status=$statuses")
}
}

171
src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt Normal file
View File

@@ -0,0 +1,171 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfigReader
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.DefaultRpcClient
import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.core.env.Environment
import org.springframework.scheduling.annotation.Scheduled
import org.springframework.stereotype.Repository
import reactor.core.publisher.toFlux
import java.io.File
import java.net.URI
import java.util.*
import javax.annotation.PostConstruct
@Repository
open class ConfiguredUpstreams(
@Autowired val env: Environment,
@Autowired private val objectMapper: ObjectMapper
) : Upstreams {
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
private val chainMapping = HashMap<Chain, ChainUpstreams>()
private val chainNames = mapOf(
"ethereum" to Chain.ETHEREUM,
"ethereum-classic" to Chain.ETHEREUM_CLASSIC,
"eth" to Chain.ETHEREUM,
"etc" to Chain.ETHEREUM_CLASSIC,
"morden" to Chain.TESTNET_MORDEN,
"kovan" to Chain.TESTNET_KOVAN
)
@PostConstruct
fun start() {
val config = readConfig()
val defaultOptions = buildDefaultOptions(config)
val groups = HashMap<Chain, ArrayList<Upstream>>()
config.upstreams.forEach { up ->
if (up.provider == "dshackle") {
buildGrpcUpstream(up)
} else {
buildEthereumUpstream(up, defaultOptions, groups)
}
}
groups.forEach { chain, group ->
chainMapping[chain] = ChainUpstreams(chain, group)
}
}
private fun readConfig(): UpstreamsConfig {
val path = env.getProperty("upstreams.config")
if (StringUtils.isEmpty(path)) {
log.error("Path to upstreams is not set (upstreams.config)")
System.exit(1)
}
val upstreamConfig = File(path!!)
val ok = upstreamConfig.exists() && upstreamConfig.isFile
if (!ok) {
log.error("Unable to setup upstreams from ${upstreamConfig.path}")
System.exit(1)
}
log.info("Read upstream configuration from ${upstreamConfig.path}")
val reader = UpstreamsConfigReader()
return reader.read(upstreamConfig.inputStream())
}
private fun buildDefaultOptions(config: UpstreamsConfig): HashMap<Chain, UpstreamsConfig.Options> {
val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
config.defaultOptions.forEach { df ->
df.chains.forEach { chainName ->
chainNames[chainName]?.let { chain ->
var current = defaultOptions[chain]
if (current == null) {
current = df.options
} else {
current = current.merge(df.options)
}
defaultOptions[chain] = current
}
}
}
return defaultOptions
}
private fun buildEthereumUpstream(up: UpstreamsConfig.Upstream,
defaultOptions: HashMap<Chain, UpstreamsConfig.Options>,
groups: HashMap<Chain, ArrayList<Upstream>>) {
val chain = chainNames[up.chain] ?: return
var rpcApi: EthereumApi? = null
var wsApi: EthereumWs? = null
val urls = ArrayList<URI>()
up.endpoints.forEach { endpoint ->
if (endpoint.type == UpstreamsConfig.EndpointType.JSON_RPC) {
rpcApi = EthereumApi(
DefaultRpcClient(DefaultRpcTransport(endpoint.url)),
objectMapper,
chain
)
}
if (endpoint.type == UpstreamsConfig.EndpointType.WEBSOCKET) {
wsApi = EthereumWs(
endpoint.url,
endpoint.origin ?: URI("http://localhost")
)
wsApi!!.connect()
}
urls.add(endpoint.url)
}
val options = (up.options ?: UpstreamsConfig.Options())
.merge(defaultOptions[chain])
.merge(UpstreamsConfig.Options.getDefaults())
if (rpcApi != null) {
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val current = groups[chain] ?: ArrayList()
current.add(EthereumUpstream(chain, rpcApi!!, wsApi, options))
groups[chain] = current
}
}
private fun buildGrpcUpstream(up: UpstreamsConfig.Upstream) {
if (up.endpoints.size == 0) {
return
}
val options = (up.options ?: UpstreamsConfig.Options())
.merge(UpstreamsConfig.Options.getDefaults())
val endpoint = up.endpoints.first()
if (endpoint.type == UpstreamsConfig.EndpointType.DSHACKLE) {
val ds = GrpcUpstreams(
endpoint.host,
endpoint.port ?: 443,
objectMapper,
options
)
log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
ds.start()
.flatMapMany {
it.toFlux()
}
.subscribe {
log.info("Subscribed to $it through gRPC at ${endpoint.host}:${endpoint.port}")
ethereumUpstream(it).addUpstream(ds.getOrCreate(it))
}
}
}
override fun ethereumUpstream(chain: Chain): ChainUpstreams {
val current = chainMapping[chain]
if (current == null) {
val created = ChainUpstreams(chain, ArrayList<Upstream>())
chainMapping[chain] = created
return created
}
return current
}
@Scheduled(fixedRate = 15000)
fun printStatuses() {
chainMapping.forEach { it.value.printStatus() }
}
override fun getAvailable(): List<Chain> {
return Collections.unmodifiableList(chainMapping.keys.toList())
}
}

27
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt
View File

@@ -3,24 +3,25 @@ package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.RpcCall
import io.infinitape.etherjar.rpc.RpcClient
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.json.ResponseJson
import io.infinitape.etherjar.rpc.transport.BatchStatus
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.toFlux
import java.time.Duration
import java.util.*
import java.util.concurrent.CompletableFuture
class EthereumApi(
private val rpcClient: RpcClient,
open class EthereumApi(
val rpcClient: RpcClient,
private val objectMapper: ObjectMapper,
private val chain: Chain
) {
private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
private val timeout = Duration.ofSeconds(5)
private val log = LoggerFactory.getLogger(EthereumApi::class.java)
var ws: EthereumWs? = null
@@ -65,13 +66,19 @@ class EthereumApi(
"eth_accounts"
)
fun execute(batch: Batch): CompletableFuture<BatchStatus> {
return rpcClient.execute(batch)
open fun <JS, RS> executeAndConvert(rpcCall: RpcCall<JS, RS>): Mono<RS> {
return execute(0, rpcCall.method, rpcCall.params as List<Any>)
.map {
jacksonRpcConverter.fromJson(it.inputStream(), rpcCall.jsonType, Int::class.java)
}.map {
rpcCall.converter.apply(it)
}
}
fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
open fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
val result: Mono<Any> = if (hardcodedMethods.contains(method)) {
Mono.just(method).map { hardcoded(it) }
Mono.just(method)
.map { hardcoded(it) }
} else if (allowedMethods.contains(method)) {
callUpstream(method, params)
} else {

102
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt Normal file
View File

@@ -0,0 +1,102 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.transport.BatchStatus
import io.infinitape.etherjar.rpc.transport.RpcTransport
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.toFlux
import reactor.util.function.Tuple3
import reactor.util.function.Tuples
import java.util.concurrent.CompletableFuture
import java.util.function.Function
class EthereumGrpcTransport(
private val chain: Chain,
private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val objectMapper: ObjectMapper
): RpcTransport {
private val chainRef = Common.ChainRef.forNumber(chain.id)
private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
override fun close() {
}
private fun replyProcessor(mapping: HashMap<Int, Batch.BatchItem<Any, Any>>): Function<BlockchainOuterClass.NativeCallReplyItem, Boolean> {
return Function { resp ->
val id = resp.id
val bi = mapping.remove(id)
if (bi != null) {
if (resp.succeed) {
try {
val rpcResp = jacksonRpcConverter.fromJson(resp.payload.toByteArray().inputStream(), bi.call.jsonType, Int::class.java)
bi.onComplete(rpcResp)
return@Function true
} catch (e: RpcException) {
bi.onError(e)
}
} else {
bi.onError(RpcException(-32603, resp.error.toString()))
}
}
false
}
}
private val sumStatus = { t: Tuple3<Int, Int, Int>, ok: Boolean ->
if (ok) Tuples.of(t.t1 + 1, t.t2, t.t3 + 1)
else Tuples.of(t.t1, t.t2 + 1, t.t3 + 1)
}
private val asStatus = Function<Tuple3<Int, Int, Int>, BatchStatus> {
BatchStatus.newBuilder()
.withSucceed(it.t1)
.withFailed(it.t2)
.withTotal(it.t3)
.build()
}
fun prepareMapping(items: List<Batch.BatchItem<out Any, out Any>>, req: BlockchainOuterClass.NativeCallRequest.Builder): HashMap<Int, Batch.BatchItem<Any, Any>> {
val mapping = HashMap<Int, Batch.BatchItem<Any, Any>>()
var seq: Int = 0
items.forEach { bi ->
val id = seq++
mapping[id] = bi as Batch.BatchItem<Any, Any>
val call = bi.call
val params = objectMapper.writeValueAsBytes(call.params)
val nativeCallItem = BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(id)
.setMethod("POST")
.setTarget(call.method)
.setPayload(ByteString.copyFrom(params))
.build()
req.addItems(nativeCallItem)
}
return mapping
}
override fun execute(items: List<Batch.BatchItem<out Any, out Any>>): CompletableFuture<BatchStatus> {
val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(chainRef);
val mapping = prepareMapping(items, req)
return client.nativeCall(req.build())
.map(replyProcessor(mapping))
.reduce(Tuples.of(0, 0, 0), sumStatus)
.map(asStatus)
.doFinally {
mapping.values.forEach { bi ->
bi.onError(RpcException(-32603, "RPC response not received"))
}
}
.toFuture()
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt
View File

@@ -25,13 +25,13 @@ class EthereumRpcHead(
.flatMap {
val batch = Batch()
val f = batch.add(Commands.eth().blockNumber)
api.execute(batch)
api.rpcClient.execute(batch)
Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5))
}
.flatMap {
val batch = Batch()
val f = batch.add(Commands.eth().getBlock(it))
api.execute(batch)
api.rpcClient.execute(batch)
Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5))
}
.onErrorContinue { err, _ ->

56
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt Normal file
View File

@@ -0,0 +1,56 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicReference
class EthereumUpstream(
val chain: Chain,
private val api: EthereumApi,
private val ethereumWs: EthereumWs? = null,
private val options: UpstreamsConfig.Options
): Upstream {
private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
private val head: EthereumHead = if (ethereumWs != null) {
EthereumWsHead(ethereumWs)
} else {
EthereumRpcHead(api).apply {
this.start()
}
}
private val validator = UpstreamValidator(this, options)
private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE)
init {
log.info("Configured for ${chain.chainName}")
validator.start()
.subscribe {
status.set(it)
}
}
override fun isAvailable(): Boolean {
return status.get() == UpstreamAvailability.OK
}
override fun getStatus(): UpstreamAvailability {
return status.get()
}
override fun getHead(): EthereumHead {
return head
}
override fun getApi(): EthereumApi {
return api
}
override fun getOptions(): UpstreamsConfig.Options {
return options
}
}

143
src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt Normal file
View File

@@ -0,0 +1,143 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import com.salesforce.reactorgrpc.GrpcRetry
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.config.UpstreamsConfig
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.json.BlockJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.TopicProcessor
import reactor.core.publisher.toMono
import java.math.BigInteger
import java.time.Duration
import java.util.concurrent.atomic.AtomicReference
import java.util.function.Function
open class GrpcUpstream(
private val chain: Chain,
private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val objectMapper: ObjectMapper,
private val options: UpstreamsConfig.Options
): Upstream {
constructor(chain: Chain, client: ReactorBlockchainGrpc.ReactorBlockchainStub, objectMapper: ObjectMapper)
: this(chain, client, objectMapper, UpstreamsConfig.Options.getDefaults())
private val log = LoggerFactory.getLogger(GrpcUpstream::class.java)
private val headBlock = AtomicReference<BlockJson<TransactionId>>(null)
private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
private var status = AtomicReference<UpstreamAvailability>(UpstreamAvailability.UNAVAILABLE)
private val head = Head(this)
private val api: EthereumApi
init {
val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
val rpcClient = DefaultRpcClient(grpcTransport)
api = EthereumApi(rpcClient, objectMapper, chain)
}
open fun connect() {
val chainRef = Common.Chain.newBuilder()
.setTypeValue(chain.id)
.build()
.toMono()
val retry: Function<Flux<BlockchainOuterClass.ChainHead>, Flux<BlockchainOuterClass.ChainHead>> = Function {
status.set(UpstreamAvailability.UNAVAILABLE)
client.subscribeHead(chainRef)
}
val flux = client.subscribeHead(chainRef)
.compose(GrpcRetry.ManyToMany.retryAfter(retry, Duration.ofSeconds(5)))
subscribe(flux)
}
internal fun subscribe(flux: Flux<BlockchainOuterClass.ChainHead>) {
flux.map { value ->
val block = BlockJson<TransactionId>()
block.number = value.height
block.totalDifficulty = BigInteger(1, value.weight.toByteArray())
block.hash = BlockHash.from("0x"+value.blockId)
block
}
.filter { block ->
val curr = headBlock.get()
curr == null || curr.totalDifficulty < block.totalDifficulty
}
.doOnError { err ->
log.error("Head subscription error", err)
}
.subscribe { block ->
log.debug("New block ${block.number} on ${chain}")
headBlock.set(block)
streamBlocks.onNext(block)
status.set(UpstreamAvailability.OK)
}
}
fun init(conf: BlockchainOuterClass.DescribeChain) {
val available = conf.available
val quorum = conf.quorum
status.set(
if (available && quorum > 0) UpstreamAvailability.OK else UpstreamAvailability.UNAVAILABLE
)
}
fun onStatus(value: BlockchainOuterClass.ChainStatus) {
val available = value.available
val quorum = value.quorum
status.set(
if (available && quorum > 0) UpstreamAvailability.OK else UpstreamAvailability.UNAVAILABLE
)
}
// ------------------------------------------------------------------------------------------
override fun isAvailable(): Boolean {
return headBlock.get() != null
}
override fun getStatus(): UpstreamAvailability {
return status.get()
}
override fun getHead(): EthereumHead {
return head
}
override fun getApi(): EthereumApi {
return api
}
override fun getOptions(): UpstreamsConfig.Options {
return options
}
class Head(
val upstream: GrpcUpstream
): EthereumHead {
override fun getHead(): Mono<BlockJson<TransactionId>> {
val current = upstream.headBlock.get()
if (current != null) {
return Mono.just(current)
}
return Mono.from(upstream.streamBlocks)
}
override fun getFlux(): Flux<BlockJson<TransactionId>> {
return Flux.from(upstream.streamBlocks)
}
}
}

69
src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt Normal file
View File

@@ -0,0 +1,69 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.grpc.Chain
import io.grpc.ManagedChannelBuilder
import reactor.core.publisher.Mono
import java.util.*
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
class GrpcUpstreams(
private val host: String,
private val port: Int,
private val objectMapper: ObjectMapper,
private val options: UpstreamsConfig.Options
) {
private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
private var known = HashMap<Chain, GrpcUpstream>()
private val lock = ReentrantLock()
fun start(): Mono<List<Chain>> {
val channel = ManagedChannelBuilder.forAddress(host, port)
.enableRetry()
channel.usePlaintext()
val client = ReactorBlockchainGrpc.newReactorStub(channel.build())
this.client = client
val loaded = client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
.map { value ->
val chains = ArrayList<Chain>()
value.chainsList.forEach { chainDetails ->
val chain = Chain.byId(chainDetails.chain.number)
if (chain != Chain.UNSPECIFIED) {
getOrCreate(chain)
.init(chainDetails)
chains.add(chain)
}
}
chains as List<Chain>
}
//TODO subscribe only after receiving details
client.subscribeStatus(BlockchainOuterClass.StatusRequest.newBuilder().build())
.subscribe { value ->
val chain = Chain.byId(value.chain.number)
if (chain != Chain.UNSPECIFIED) {
getOrCreate(chain).onStatus(value)
}
}
return loaded
}
fun getOrCreate(chain: Chain): GrpcUpstream {
lock.withLock {
val current = known[chain]
return if (current == null) {
val created = GrpcUpstream(chain, client!!, objectMapper, options)
known[chain] = created
created.connect()
created
} else {
current
}
}
}
}

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

@@ -1,44 +1,11 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicReference
class Upstream(
val chain: Chain,
val api: EthereumApi,
private val ethereumWs: EthereumWs? = null,
private val options: UpstreamsConfig.Options
) {
private val log = LoggerFactory.getLogger(Upstream::class.java)
val head: EthereumHead = if (ethereumWs != null) {
EthereumWsHead(ethereumWs)
} else {
EthereumRpcHead(api).apply {
this.start()
}
}
val validator = UpstreamValidator(this, options)
private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE)
init {
log.info("Configured for ${chain.chainName}")
validator.start()
.subscribe {
status.set(it)
}
}
fun isAvailable(): Boolean {
return status.get() == UpstreamAvailability.OK
}
fun getStatus(): UpstreamAvailability {
return status.get()
}
interface Upstream {
fun isAvailable(): Boolean
fun getStatus(): UpstreamAvailability
fun getHead(): EthereumHead
fun getApi(): EthereumApi
fun getOptions(): UpstreamsConfig.Options
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt
View File

@@ -8,7 +8,7 @@ import java.time.Duration
import java.util.concurrent.TimeUnit
class UpstreamValidator(
private val upstream: Upstream,
private val ethereumUpstream: EthereumUpstream,
private val options: UpstreamsConfig.Options
) {
@@ -17,7 +17,7 @@ class UpstreamValidator(
val peerCount = batch.add(Commands.net().peerCount())
val syncing = batch.add(Commands.eth().syncing())
try {
upstream.api.execute(batch).get(5, TimeUnit.SECONDS)
ethereumUpstream.getApi().rpcClient.execute(batch).get(5, TimeUnit.SECONDS)
if (syncing.get().isSyncing) {
return UpstreamAvailability.SYNCING
}

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

@@ -1,116 +1,8 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfigReader
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.DefaultRpcClient
import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.core.env.Environment
import org.springframework.scheduling.annotation.Scheduled
import org.springframework.stereotype.Repository
import java.io.File
import java.net.URI
import javax.annotation.PostConstruct
@Repository
class Upstreams(
@Autowired val env: Environment,
@Autowired private val objectMapper: ObjectMapper
) {
private val log = LoggerFactory.getLogger(Upstreams::class.java)
private val chainMapping = HashMap<Chain, ChainConnect>()
private val chainNames = mapOf(
"ethereum" to Chain.ETHEREUM,
"ethereum-classic" to Chain.ETHEREUM_CLASSIC,
"eth" to Chain.ETHEREUM,
"etc" to Chain.ETHEREUM_CLASSIC,
"morden" to Chain.TESTNET_MORDEN,
"kovan" to Chain.TESTNET_KOVAN
)
@PostConstruct
fun start() {
val path = env.getProperty("upstreams.config")
if (StringUtils.isEmpty(path)) {
log.error("Path to upstreams is not set (upstreams.config)")
System.exit(1)
}
val upstreamConfig = File(path)
val ok = upstreamConfig.exists() && upstreamConfig.isFile
if (!ok) {
log.error("Unable to setup upstreams from ${upstreamConfig.path}")
System.exit(1)
}
log.info("Read upstream configuration from ${upstreamConfig.path}")
val reader = UpstreamsConfigReader()
val config = reader.read(upstreamConfig.inputStream())
val groups = HashMap<Chain, ArrayList<Upstream>>()
val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
config.defaultOptions.forEach { df ->
df.chains.forEach { chainName ->
chainNames[chainName]?.let { chain ->
var current = defaultOptions[chain]
if (current == null) {
current = df.options
} else {
current = current.merge(df.options)
}
defaultOptions[chain] = current
}
}
}
config.upstreams.forEach { up ->
val chain = chainNames[up.chain] ?: return@forEach
var rpcApi: EthereumApi? = null
var wsApi: EthereumWs? = null
val urls = ArrayList<URI>()
up.endpoints.forEach { endpoint ->
if (endpoint.type == UpstreamsConfig.EndpointType.JSON_RPC) {
rpcApi = EthereumApi(
DefaultRpcClient(DefaultRpcTransport(endpoint.url)),
objectMapper,
chain
)
}
if (endpoint.type == UpstreamsConfig.EndpointType.WEBSOCKET) {
wsApi = EthereumWs(
endpoint.url,
endpoint.origin ?: URI("http://localhost")
)
wsApi!!.connect()
}
urls.add(endpoint.url)
}
val options = (up.options ?: UpstreamsConfig.Options())
.merge(defaultOptions[chain])
.merge(UpstreamsConfig.Options.getDefaults())
if (rpcApi != null) {
log.info("Info using ${chain.chainName} upstream, at ${urls.joinToString()}")
val current = groups[chain] ?: ArrayList()
current.add(Upstream(chain, rpcApi!!, wsApi, options))
groups[chain] = current
}
}
groups.forEach { chain, group ->
chainMapping[chain] = ChainConnect(chain, group)
}
}
fun ethereumUpstream(chain: Chain): ChainConnect? {
return chainMapping[chain]
}
@Scheduled(fixedRate = 15000)
fun printStatuses() {
chainMapping.forEach { it.value.printStatus() }
}
interface Upstreams {
fun ethereumUpstream(chain: Chain): AggregatedUpstreams
fun getAvailable(): List<Chain>
}

32
src/main/resources/log4j2.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Properties>
<Property name="dfltPattern">%d{HH:mm:ss.SSS} [%-20t] %-5level %24.24c{1} | %msg%n</Property>
</Properties>
<Appenders>
<Console name="STDOUT" target="SYSTEM_OUT">
<PatternLayout pattern="${dfltPattern}"/>
<Filters>
<ThresholdFilter level="WARN" onMatch="DENY" onMismatch="ACCEPT" />
</Filters>
</Console>
<Console name="STDERR" target="SYSTEM_ERR">
<PatternLayout pattern="${dfltPattern}" />
</Console>
</Appenders>
<Loggers>
<Logger name="io.emeraldpay" level="debug" additivity="false">
<AppenderRef ref="STDOUT"/>
<AppenderRef ref="STDERR" level="warn"/>
</Logger>
<Root level="warn" additivity="false">
<AppenderRef ref="STDOUT"/>
<AppenderRef ref="STDERR" level="warn"/>
</Root>
</Loggers>
</Configuration>

27
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

@@ -54,4 +54,31 @@ class UpstreamsConfigReaderSpec extends Specification {
}
}
def "Parse ds config"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-ds.yaml")
when:
def act = reader.read(config)
then:
act != null
act.version == "v1"
act.upstreams.size() == 1
with(act.upstreams.get(0)) {
id == "remote"
chain == "auto"
provider == "dshackle"
endpoints.size() == 1
with(endpoints.get(0)) {
type == UpstreamsConfig.EndpointType.DSHACKLE
host == "10.2.0.15"
auth instanceof UpstreamsConfig.TlsAuth
with((UpstreamsConfig.TlsAuth)auth) {
ca == "/etc/ca.myservice.com.crt"
certificate == "/etc/client1.myservice.com.crt"
key == "/etc/client1.myservice.com.key"
}
}
}
}
}

54
src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy Normal file
View File

@@ -0,0 +1,54 @@
package io.emeraldpay.dshackle.test
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.EthereumApi
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.RpcClient
import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.ResponseJson
import org.jetbrains.annotations.NotNull
import reactor.core.publisher.Mono
class EthereumApiMock extends EthereumApi {
List<PredefinedResponse> predefined = []
private ObjectMapper objectMapper
EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) {
super(rpcClient, objectMapper, chain)
this.objectMapper = objectMapper
}
EthereumApiMock answer(@NotNull String method, List<Object> params, Object result) {
predefined << new PredefinedResponse(method: method, params: params, result: result)
return this
}
@Override
Mono<byte[]> execute(int id, @NotNull String method, @NotNull List<?> params) {
def predefined = predefined.find { it.isSame(id, method, params) }
ResponseJson json = new ResponseJson<Object, Integer>(id: id)
if (predefined != null) {
json.result = predefined.result
} else {
json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked")
}
return Mono.just(objectMapper.writeValueAsBytes(json))
}
class PredefinedResponse {
String method
List params
Object result
boolean isSame(int id, String method, List<?> params) {
if (method != this.method) {
return false
}
if (this.params == null) {
return true
}
return this.params == params
}
}
}

20
src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy Normal file
View File

@@ -0,0 +1,20 @@
package io.emeraldpay.dshackle.test
import io.emeraldpay.api.proto.BlockchainGrpc
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.grpc.inprocess.InProcessChannelBuilder
import io.grpc.inprocess.InProcessServerBuilder
import io.grpc.testing.GrpcCleanupRule
class MockServer {
GrpcCleanupRule grpcCleanup = new GrpcCleanupRule()
ReactorBlockchainGrpc.ReactorBlockchainStub runServer(BlockchainGrpc.BlockchainImplBase impl){
String serverName = InProcessServerBuilder.generateName()
grpcCleanup.register(InProcessServerBuilder
.forName(serverName).directExecutor().addService(impl).build().start());
def channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build())
return ReactorBlockchainGrpc.newReactorStub(channel)
}
}

24
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy Normal file
View File

@@ -0,0 +1,24 @@
package io.emeraldpay.dshackle.test
import com.fasterxml.jackson.core.Version
import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.module.SimpleModule
import java.text.SimpleDateFormat
class TestingCommons {
static ObjectMapper objectMapper() {
def module = new SimpleModule("EmeraldDShackle", new Version(1, 0, 0, null, null, null))
def objectMapper = new ObjectMapper()
objectMapper.registerModule(module)
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
objectMapper
.setDateFormat(new SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS"))
.setTimeZone(TimeZone.getTimeZone("UTC"))
return objectMapper
}
}

112
src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy Normal file
View File

@@ -0,0 +1,112 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainGrpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.test.EthereumApiMock
import io.emeraldpay.dshackle.test.MockServer
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.grpc.Chain
import io.grpc.stub.StreamObserver
import io.infinitape.etherjar.rpc.Batch
import io.infinitape.etherjar.rpc.RpcCall
import io.infinitape.etherjar.rpc.RpcClient
import spock.lang.Specification
class EthereumGrpcTransportSpec extends Specification {
MockServer mockServer = new MockServer()
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Make simple call"() {
setup:
def callData = [:]
def otherSideUpstreams = Mock(Upstreams)
def otherSideAggr = Mock(AggregatedUpstreams)
def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)
def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
callData["request"] = request
otherSideNativeCall.nativeCall(request, responseObserver)
}
})
EthereumGrpcTransport transport = new EthereumGrpcTransport(Chain.ETHEREUM, client, objectMapper)
when:
otherSideApi.answer("eth_test", [1], "bar")
def batch = new Batch()
def f = batch.add(RpcCall.create("eth_test", [1]))
def status = transport.execute(batch.items).get()
then:
1 * otherSideUpstreams.ethereumUpstream(Chain.ETHEREUM) >> otherSideAggr
1 * otherSideAggr.api >> otherSideApi
status.failed == 0
status.succeed == 1
status.total == 1
callData.request != null
with((BlockchainOuterClass.NativeCallRequest)callData.request) {
chain.number == Chain.ETHEREUM.id
itemsCount == 1
with(getItems(0)) {
target == "eth_test"
payload.toStringUtf8() == "[1]"
}
}
f.get() == "bar"
}
def "Make few calls"() {
setup:
def callData = [:]
def otherSideUpstreams = Mock(Upstreams)
def otherSideAggr = Mock(AggregatedUpstreams)
def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)
def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
callData["request"] = request
otherSideNativeCall.nativeCall(request, responseObserver)
}
})
EthereumGrpcTransport transport = new EthereumGrpcTransport(Chain.ETHEREUM, client, objectMapper)
when:
otherSideApi.answer("eth_test", [1], "bar")
otherSideApi.answer("eth_test2", [2, "3"], "baz")
def batch = new Batch()
def f1 = batch.add(RpcCall.create("eth_test", [1]))
def f2 = batch.add(RpcCall.create("eth_test2", [2, "3"]))
def status = transport.execute(batch.items).get()
then:
1 * otherSideUpstreams.ethereumUpstream(Chain.ETHEREUM) >> otherSideAggr
1 * otherSideAggr.api >> otherSideApi
status.failed == 0
status.succeed == 2
status.total == 2
callData.request != null
with((BlockchainOuterClass.NativeCallRequest)callData.request) {
chain.number == Chain.ETHEREUM.id
itemsCount == 2
with(getItems(0)) {
target == "eth_test"
payload.toStringUtf8() == "[1]"
}
with(getItems(1)) {
target == "eth_test2"
payload.toStringUtf8() == "[2,\"3\"]"
}
}
f1.get() == "bar"
f2.get() == "baz"
}
}

121
src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy Normal file
View File

@@ -0,0 +1,121 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainGrpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.test.MockServer
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.grpc.Chain
import io.grpc.stub.StreamObserver
import io.infinitape.etherjar.domain.BlockHash
import org.apache.commons.codec.binary.Hex
import spock.lang.Specification
import java.util.concurrent.CompletableFuture
class GrpcUpstreamSpec extends Specification {
MockServer mockServer = new MockServer()
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Subscribe to head"() {
setup:
def callData = [:]
def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
callData.chain = request.getTypeValue()
responseObserver.onNext(
BlockchainOuterClass.ChainHead.newBuilder()
.setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
.setHeight(650246)
.setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456")))
.build()
)
}
})
def chain = Chain.ETHEREUM
def upstream = new GrpcUpstream(chain, client, objectMapper)
when:
upstream.connect()
def h = upstream.head.head.block()
then:
callData.chain == Chain.ETHEREUM.id
upstream.status == UpstreamAvailability.OK
h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
}
def "Follows difficulty, ignores less difficult"() {
setup:
def callData = [:]
def finished = new CompletableFuture<Boolean>()
def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
responseObserver.onNext(
BlockchainOuterClass.ChainHead.newBuilder()
.setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
.setHeight(650246)
.setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456")))
.build()
)
responseObserver.onNext(
BlockchainOuterClass.ChainHead.newBuilder()
.setBlockId("3ec2ebf5d0ec474d0ac6bca770d8409ad750d26e119968e7919f85d5ec891521")
.setHeight(650247)
.setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6455")))
.build()
)
finished.complete(true)
}
})
def chain = Chain.ETHEREUM
def upstream = new GrpcUpstream(chain, client, objectMapper)
when:
upstream.connect()
finished.get()
def h = upstream.head.head.block()
then:
upstream.status == UpstreamAvailability.OK
h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
h.number == 650246
}
def "Follows difficulty"() {
setup:
def callData = [:]
def finished = new CompletableFuture<Boolean>()
def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
responseObserver.onNext(
BlockchainOuterClass.ChainHead.newBuilder()
.setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
.setHeight(650246)
.setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456")))
.build()
)
responseObserver.onNext(
BlockchainOuterClass.ChainHead.newBuilder()
.setBlockId("3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")
.setHeight(650247)
.setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6457")))
.build()
)
finished.complete(true)
}
})
def chain = Chain.ETHEREUM
def upstream = new GrpcUpstream(chain, client, objectMapper)
when:
upstream.connect()
finished.get()
def h = upstream.head.head.block()
then:
upstream.status == UpstreamAvailability.OK
h.hash == BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")
h.number == 650247
}
}

32
src/test/resources/log4j2.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Properties>
<Property name="dfltPattern">%d{HH:mm:ss.SSS} [%-20t] %-5level %24.24c{1} | %msg%n</Property>
</Properties>
<Appenders>
<Console name="STDOUT" target="SYSTEM_OUT">
<PatternLayout pattern="${dfltPattern}"/>
<Filters>
<ThresholdFilter level="WARN" onMatch="DENY" onMismatch="ACCEPT" />
</Filters>
</Console>
<Console name="STDERR" target="SYSTEM_ERR">
<PatternLayout pattern="${dfltPattern}" />
</Console>
</Appenders>
<Loggers>
<Logger name="io.emeraldpay" level="debug" additivity="false">
<AppenderRef ref="STDOUT"/>
<AppenderRef ref="STDERR" level="warn"/>
</Logger>
<Root level="warn" additivity="false">
<AppenderRef ref="STDOUT"/>
<AppenderRef ref="STDERR" level="warn"/>
</Root>
</Loggers>
</Configuration>