From fd8c35bec2cb5049e4c16b461f97da2026c0968e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 25 Jul 2019 22:53:02 -0400 Subject: [PATCH] solution: connects to remote dshackle --- .gitignore | 3 +- build.gradle | 5 +- .../dshackle/config/UpstreamsConfig.java | 111 ++++++++++-- .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 12 +- .../io/emeraldpay/dshackle/rpc/Describe.kt | 43 +++++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 3 +- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 8 +- .../dshackle/rpc/SubscribeStatus.kt | 19 ++ .../emeraldpay/dshackle/rpc/TrackAddress.kt | 30 +-- .../io/emeraldpay/dshackle/rpc/TrackTx.kt | 24 +-- .../dshackle/upstream/AggregatedUpstreams.kt | 52 ++++++ .../dshackle/upstream/ChainConnect.kt | 93 ---------- .../dshackle/upstream/ChainUpstreams.kt | 72 ++++++++ .../dshackle/upstream/ConfiguredUpstreams.kt | 171 ++++++++++++++++++ .../dshackle/upstream/EthereumApi.kt | 27 ++- .../upstream/EthereumGrpcTransport.kt | 102 +++++++++++ .../dshackle/upstream/EthereumRpcHead.kt | 4 +- .../dshackle/upstream/EthereumUpstream.kt | 56 ++++++ .../dshackle/upstream/GrpcUpstream.kt | 143 +++++++++++++++ .../dshackle/upstream/GrpcUpstreams.kt | 69 +++++++ .../emeraldpay/dshackle/upstream/Upstream.kt | 45 +---- .../dshackle/upstream/UpstreamValidator.kt | 4 +- .../emeraldpay/dshackle/upstream/Upstreams.kt | 114 +----------- src/main/resources/log4j2.xml | 32 ++++ .../config/UpstreamsConfigReaderSpec.groovy | 27 +++ .../dshackle/test/EthereumApiMock.groovy | 54 ++++++ .../dshackle/test/MockServer.groovy | 20 ++ .../dshackle/test/TestingCommons.groovy | 24 +++ .../upstream/EthereumGrpcTransportSpec.groovy | 112 ++++++++++++ .../dshackle/upstream/GrpcUpstreamSpec.groovy | 121 +++++++++++++ src/test/resources/log4j2.xml | 32 ++++ 31 files changed, 1320 insertions(+), 312 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt create mode 100644 src/main/resources/log4j2.xml create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy create mode 100644 src/test/resources/log4j2.xml diff --git a/.gitignore b/.gitignore index 2d7aca60..cd79f4b0 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,5 @@ build/ out/ *.iml -*.yaml \ No newline at end of file +*.yaml +testsetup/ \ No newline at end of file diff --git a/build.gradle b/build.gradle index d701e584..5aa8a838 100644 --- a/build.gradle +++ b/build.gradle @@ -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 { diff --git a/src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java b/src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java index e36d6bb9..c0987e7a 100644 --- a/src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java +++ b/src/main/java/io/emeraldpay/dshackle/config/UpstreamsConfig.java @@ -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 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 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 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()); } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index c8346ba3..37ca022a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -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) { + describe.describe(request, responseObserver) + } + + override fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver) { + subscribeStatus.subscribeStatus(request, responseObserver) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt new file mode 100644 index 00000000..cb2bed9b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -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) { + 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() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index dddd5fbe..bbd09ac2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 9d5f2320..559841e9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -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): 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()!! diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt new file mode 100644 index 00000000..df1cef75 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -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) { + upstreams.getAvailable().forEach { chain -> + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index 48fda2c0..be5ab808 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -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): Flux { - val up = upstreams.ethereumUpstream(chain)!! + val up = upstreams.ethereumUpstream(chain) return group.toFlux() - .reduce>>(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) + class Update(val addr: TrackedAddress, val value: Wei) class TrackedAddress(val chain: Chain, val stream: StreamSender, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index aeb93dda..110a3263 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -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 { - 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 ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt new file mode 100644 index 00000000..fa1d6cef --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt @@ -0,0 +1,52 @@ +package io.emeraldpay.dshackle.upstream + +abstract class AggregatedUpstreams { + + abstract fun getAll(): List + abstract fun addUpstream(upstream: Upstream) + abstract fun getApis(quorum: Int): Iterator + abstract fun getApi(): EthereumApi + abstract fun getHead(): EthereumHead + + class SingleApi( + private val quorumApi: QuorumApi + ): Iterator { + + 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, + private val quorum: Int, + private var pos: Int + ): Iterator { + + 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") + } + + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt deleted file mode 100644 index 0434e72a..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainConnect.kt +++ /dev/null @@ -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 - ) { - - 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 { - 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 { - - 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, - private val quorum: Int, - private var pos: Int - ): Iterator { - - 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") - } - - } - - -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt new file mode 100644 index 00000000..2cd11486 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -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 +) : 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 { + return upstreams + } + + override fun addUpstream(upstream: Upstream) { + upstreams.add(upstream) + head = updateHead() + } + + override fun getApis(quorum: Int): Iterator { + 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") + } + + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt new file mode 100644 index 00000000..a9776d9a --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -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() + + 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>() + 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 { + val defaultOptions = HashMap() + 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, + groups: HashMap>) { + val chain = chainNames[up.chain] ?: return + var rpcApi: EthereumApi? = null + var wsApi: EthereumWs? = null + val urls = ArrayList() + 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()) + chainMapping[chain] = created + return created + } + return current + } + + @Scheduled(fixedRate = 15000) + fun printStatuses() { + chainMapping.forEach { it.value.printStatus() } + } + + override fun getAvailable(): List { + return Collections.unmodifiableList(chainMapping.keys.toList()) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index 5914347f..6e86012b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -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 { - return rpcClient.execute(batch) + open fun executeAndConvert(rpcCall: RpcCall): Mono { + return execute(0, rpcCall.method, rpcCall.params as List) + .map { + jacksonRpcConverter.fromJson(it.inputStream(), rpcCall.jsonType, Int::class.java) + }.map { + rpcCall.converter.apply(it) + } } - fun execute(id: Int, method: String, params: List): Mono { + open fun execute(id: Int, method: String, params: List): Mono { val result: Mono = 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt new file mode 100644 index 00000000..109bc8f8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt @@ -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>): Function { + 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, 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, BatchStatus> { + BatchStatus.newBuilder() + .withSucceed(it.t1) + .withFailed(it.t2) + .withTotal(it.t3) + .build() + } + + fun prepareMapping(items: List>, req: BlockchainOuterClass.NativeCallRequest.Builder): HashMap> { + val mapping = HashMap>() + var seq: Int = 0 + items.forEach { bi -> + val id = seq++ + mapping[id] = bi as Batch.BatchItem + 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>): CompletableFuture { + 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() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt index 528824ac..61bf1f03 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt @@ -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, _ -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt new file mode 100644 index 00000000..aca88ed3 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -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 + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt new file mode 100644 index 00000000..625d2493 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -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>(null) + private val streamBlocks: TopicProcessor> = TopicProcessor.create() + private var status = AtomicReference(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> = 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) { + flux.map { value -> + val block = BlockJson() + 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> { + val current = upstream.headBlock.get() + if (current != null) { + return Mono.just(current) + } + return Mono.from(upstream.streamBlocks) + } + + override fun getFlux(): Flux> { + return Flux.from(upstream.streamBlocks) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt new file mode 100644 index 00000000..6e8d668d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt @@ -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() + private val lock = ReentrantLock() + + fun start(): Mono> { + 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() + 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 + } + //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 + } + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 01cf6e69..b84b2e1c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -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 } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt index 1c9b530e..47f52422 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamValidator.kt @@ -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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index 0499c86b..5bec84d0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -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() - - 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>() - - val defaultOptions = HashMap() - 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() - 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 } \ No newline at end of file diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml new file mode 100644 index 00000000..aa4a58a8 --- /dev/null +++ b/src/main/resources/log4j2.xml @@ -0,0 +1,32 @@ + + + + + %d{HH:mm:ss.SSS} [%-20t] %-5level %24.24c{1} | %msg%n + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 1d038d37..ea79b91b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -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" + } + } + } + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy new file mode 100644 index 00000000..817620d2 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -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 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 params, Object result) { + predefined << new PredefinedResponse(method: method, params: params, result: result) + return this + } + + @Override + Mono execute(int id, @NotNull String method, @NotNull List params) { + def predefined = predefined.find { it.isSame(id, method, params) } + ResponseJson json = new ResponseJson(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 + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy new file mode 100644 index 00000000..16dbdb4b --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy @@ -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) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy new file mode 100644 index 00000000..c3d4cf0c --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -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 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy new file mode 100644 index 00000000..db51452c --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -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 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 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" + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy new file mode 100644 index 00000000..9576e0ab --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy @@ -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 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() + def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void subscribeHead(Common.Chain request, StreamObserver 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() + def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void subscribeHead(Common.Chain request, StreamObserver 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 + } +} diff --git a/src/test/resources/log4j2.xml b/src/test/resources/log4j2.xml new file mode 100644 index 00000000..aa4a58a8 --- /dev/null +++ b/src/test/resources/log4j2.xml @@ -0,0 +1,32 @@ + + + + + %d{HH:mm:ss.SSS} [%-20t] %-5level %24.24c{1} | %msg%n + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file